Repository: getsentry/sentry
Pull request: feat(metrics) Create Investigation bias creation endpoint
This PR creates an endpoint that allows creating rules from a random query.
The idea being that a user can create a rule that will exist for a limited time 
that will collect transactions with some custom conditions. 

Resolves https://github.com/getsentry/sentry/issues/55679

Problem statement:
Endpoint - Create rule from query
This is going to create a rule and save it in redis with expiration time of 1 hour.

If we can't create a rule for this specific query, let the frontend know that we can't do it and why.

# Details

#### URL
`POST /api/0/organizations/<org_slug>/dynamic-sampling/custom-rules`
#### Request body

Request will be a Json object with the following fields
- `query`: the query (same as  `OrganizationEventsEndpoint` )
- `projects`: the projects for which to query, if omitted all ( supports multiple selection)
- `period`: str the period to collect the transaction e.g. 10m, 1d, 6h

```js
{
	projects?: int[], // optional list of projects (all must be in the current org)
	query: str,   // same as Discover i.e. OrganizationEventsEndpoint
	period?: str, // optional d(days),h(hours),m(minutes),s(seconds),w(weeks)
}
```
### Return

Response  a JSON object:
#### Success
```js
{
     "ruleId": 123,
     "condition": { "op" : "glob", "name": "releases", "value":["1.1.1", "1.1.2"]},
     "startDate": "2023-09-18T10:00:00.1234Z",
     "endDate": "2023-09-18T11:00:00.1234Z",
     "numSamples": 100,
     "sampleRate": 1.0,
     "dateAdded": "2023-09-18T10:00:00.1234Z",
     "projects": [1,2,3],
     "orgId": 44,
}
```
#### Error
django rest framework error ( i.e. `{ 'field1-name': ['error1','err2'], 'field2-name': ['err1','err2']}` )

Hints from issue discussion:

Commit message: add import/export model test.

Patch to review:
diff --git a/fixtures/backup/model_dependencies/detailed.json b/fixtures/backup/model_dependencies/detailed.json
index 7770ad00b63062..55987e7b480c19 100644
--- a/fixtures/backup/model_dependencies/detailed.json
+++ b/fixtures/backup/model_dependencies/detailed.json
@@ -609,6 +609,40 @@
       "Region"
     ]
   },
+  "sentry.CustomDynamicSamplingRule": {
+    "foreign_keys": {
+      "organization": {
+        "kind": "FlexibleForeignKey",
+        "model": "sentry.Organization"
+      }
+    },
+    "model": "sentry.CustomDynamicSamplingRule",
+    "relocation_scope": "Organization",
+    "silos": [
+      "Region"
+    ]
+  },
+  "sentry.CustomDynamicSamplingRuleProject": {
+    "foreign_keys": {
+      "custom_dynamic_sampling_rule": {
+        "kind": "FlexibleForeignKey",
+        "model": "sentry.CustomDynamicSamplingRule"
+      },
+      "organization": {
+        "kind": "FlexibleForeignKey",
+        "model": "sentry.Organization"
+      },
+      "project": {
+        "kind": "FlexibleForeignKey",
+        "model": "sentry.Project"
+      }
+    },
+    "model": "sentry.CustomDynamicSamplingRuleProject",
+    "relocation_scope": "Organization",
+    "silos": [
+      "Region"
+    ]
+  },
   "sentry.Dashboard": {
     "foreign_keys": {
       "created_by_id": {
diff --git a/fixtures/backup/model_dependencies/flat.json b/fixtures/backup/model_dependencies/flat.json
index db87c81b039283..c81193aa7b6ff6 100644
--- a/fixtures/backup/model_dependencies/flat.json
+++ b/fixtures/backup/model_dependencies/flat.json
@@ -131,6 +131,14 @@
   "sentry.Counter": [
     "sentry.Project"
   ],
+  "sentry.CustomDynamicSamplingRule": [
+    "sentry.Organization"
+  ],
+  "sentry.CustomDynamicSamplingRuleProject": [
+    "sentry.CustomDynamicSamplingRule",
+    "sentry.Organization",
+    "sentry.Project"
+  ],
   "sentry.Dashboard": [
     "sentry.Organization",
     "sentry.User"
diff --git a/fixtures/backup/model_dependencies/sorted.json b/fixtures/backup/model_dependencies/sorted.json
index 26a330cf4c35ba..e86bbfa1bfb637 100644
--- a/fixtures/backup/model_dependencies/sorted.json
+++ b/fixtures/backup/model_dependencies/sorted.json
@@ -78,6 +78,8 @@
   "sentry.EventAttachment",
   "sentry.Environment",
   "sentry.EnvironmentProject",
+  "sentry.CustomDynamicSamplingRule",
+  "sentry.CustomDynamicSamplingRuleProject",
   "sentry.Distribution",
   "sentry.Deploy",
   "sentry.DeletedTeam",
diff --git a/migrations_lockfile.txt b/migrations_lockfile.txt
index bd7d6c7bde642a..4356c1bf8e9fd3 100644
--- a/migrations_lockfile.txt
+++ b/migrations_lockfile.txt
@@ -8,5 +8,5 @@ will then be regenerated, and you should be able to merge without conflicts.
 feedback: 0002_feedback_add_org_id_and_rename_event_id
 nodestore: 0002_nodestore_no_dictfield
 replays: 0003_add_size_to_recording_segment
-sentry: 0554_add_team_replica
+sentry: 0555_custom_dynamic_sampling_rule
 social_auth: 0002_default_auto_field
diff --git a/src/sentry/api/endpoints/custom_rules.py b/src/sentry/api/endpoints/custom_rules.py
new file mode 100644
index 00000000000000..8577715bb35549
--- /dev/null
+++ b/src/sentry/api/endpoints/custom_rules.py
@@ -0,0 +1,140 @@
+from datetime import datetime, timedelta, timezone
+from typing import cast
+
+from django.db import DatabaseError
+from rest_framework import serializers
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import features
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import region_silo_endpoint
+from sentry.api.bases import OrganizationEndpoint
+from sentry.api.event_search import parse_search_query
+from sentry.exceptions import InvalidSearchQuery
+from sentry.models import CustomDynamicSamplingRule, TooManyRules
+from sentry.models.organization import Organization
+from sentry.snuba.metrics.extraction import SearchQueryConverter
+from sentry.utils import json
+from sentry.utils.dates import parse_stats_period
+
+MAX_RULE_PERIOD_STRING = "6h"
+MAX_RULE_PERIOD = parse_stats_period(MAX_RULE_PERIOD_STRING)
+DEFAULT_PERIOD_STRING = "1h"
+# the number of samples to collect per custom rule
+NUM_SAMPLES_PER_CUSTOM_RULE = 100
+_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+
+
+class CustomRulesInputSerializer(serializers.Serializer):
+    """
+    Request data serializer
+    """
+
+    # the query string in the same format as the Discover query
+    query = serializers.CharField(required=True)
+    # desired time period for collection (it may be overriden if too long)
+    period = serializers.CharField(required=False)
+    # list of project ids to collect data from
+    projects = serializers.ListField(child=serializers.IntegerField(), required=False)
+
+    def validate(self, data):
+        """
+        Optional fields cannot be validated with validate_<FIELD_NAME> so we need to do it here for the
+        `period` & 'projects' & overrideExisting fields
+        """
+        if data.get("projects") is None:
+            data["projects"] = []
+
+        period = data.get("period")
+        if period is None:
+            data["period"] = DEFAULT_PERIOD_STRING
+        else:
+            try:
+                period = parse_stats_period(period)
+            except OverflowError:
+                data["period"] = MAX_RULE_PERIOD_STRING
+            if period is None:
+                raise serializers.ValidationError("Invalid period")
+            if period > MAX_RULE_PERIOD:
+                # limit the expiry period
+                data["period"] = MAX_RULE_PERIOD_STRING
+        return data
+
+
+@region_silo_endpoint
+class CustomRulesEndpoint(OrganizationEndpoint):
+    owner = ApiOwner.TELEMETRY_EXPERIENCE
+
+    publish_status = {
+        "POST": ApiPublishStatus.EXPERIMENTAL,
+    }
+
+    def post(self, request: Request, organization: Organization) -> Response:
+
+        if not features.has("organizations:investigation-bias", organization, actor=request.user):
+            return Response(status=404)
+
+        serializer = CustomRulesInputSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=400)
+
+        query = serializer.validated_data["query"]
+        projects = serializer.validated_data.get("projects")
+        period = serializer.validated_data.get("period")
+
+        try:
+            tokens = parse_search_query(query)
+        except InvalidSearchQuery as e:
+            return Response({"query": [str(e)]}, status=400)
+
+        try:
+            converter = SearchQueryConverter(tokens)
+            condition = converter.convert()
+
+            # the parsing must succeed (it passed validation)
+            delta = cast(timedelta, parse_stats_period(period))
+            now = datetime.now(tz=timezone.utc)
+            start = now
+            end = now + delta
+
+            rule = CustomDynamicSamplingRule.update_or_create(
+                condition=condition,
+                start=start,
+                end=end,
+                project_ids=projects,
+                organization_id=organization.id,
+                num_samples=NUM_SAMPLES_PER_CUSTOM_RULE,
+                sample_rate=1.0,
+            )
+
+            response_data = {
+                "ruleId": rule.external_rule_id,
+                "condition": json.loads(rule.condition),
+                "startDate": rule.start_date.strftime(_DATE_FORMAT),
+                "endDate": rule.end_date.strftime(_DATE_FORMAT),
+                "numSamples": rule.num_samples,
+                "sampleRate": rule.sample_rate,
+                "dateAdded": rule.date_added.strftime(_DATE_FORMAT),
+                "projects": [project.id for project in rule.projects.all()],
+                "orgId": rule.organization_id,
+            }
+            return Response(response_data, status=200)
+
+        except DatabaseError:
+            return Response(
+                {"projects": ["Could not save rule, probably wrong project ids"]}, status=400
+            )
+        except TooManyRules:
+            return Response(
+                {
+                    "error": [
+                        "Too many investigation rules active for this organization."
+                        "Wait until some expire or delete some rules."
+                    ]
+                },
+                status=429,
+            )
+        except ValueError as e:
+            return Response({"query": ["Could not convert to rule", str(e)]}, status=400)
diff --git a/src/sentry/api/urls.py b/src/sentry/api/urls.py
index 0eda22a4ba223f..bba9d797fe1144 100644
--- a/src/sentry/api/urls.py
+++ b/src/sentry/api/urls.py
@@ -161,6 +161,7 @@
     ProjectCodeOwnersDetailsEndpoint,
     ProjectCodeOwnersEndpoint,
 )
+from .endpoints.custom_rules import CustomRulesEndpoint
 from .endpoints.data_scrubbing_selector_suggestions import DataScrubbingSelectorSuggestionsEndpoint
 from .endpoints.debug_files import (
     AssociateDSymFilesEndpoint,
@@ -1883,6 +1884,18 @@
             ],
         ),
     ),
+    re_path(
+        r"^(?P<organization_slug>[^/]+)/dynamic-sampling/",
+        include(
+            [
+                re_path(
+                    r"^custom-rules/$",
+                    CustomRulesEndpoint.as_view(),
+                    name="sentry-api-0-organization-dynamic_sampling-custom_rules",
+                ),
+            ],
+        ),
+    ),
 ]
 
 PROJECT_URLS: list[URLPattern | URLResolver] = [
diff --git a/src/sentry/dynamic_sampling/rules/utils.py b/src/sentry/dynamic_sampling/rules/utils.py
index 30313da5fbc385..10aca39108030e 100644
--- a/src/sentry/dynamic_sampling/rules/utils.py
+++ b/src/sentry/dynamic_sampling/rules/utils.py
@@ -3,6 +3,7 @@
 
 from django.conf import settings
 
+from sentry.models import CUSTOM_RULE_START
 from sentry.utils import json, redis
 
 BOOSTED_RELEASES_LIMIT = 10
@@ -12,7 +13,6 @@
 
 IGNORE_HEALTH_CHECKS_FACTOR = 5
 
-
 ProjectId = int
 DecisionDropCount = int
 DecisionKeepCount = int
@@ -41,6 +41,7 @@ class RuleType(Enum):
     BOOST_KEY_TRANSACTIONS_RULE = "boostKeyTransactions"
     BOOST_LOW_VOLUME_TRANSACTIONS_RULE = "boostLowVolumeTransactions"
     BOOST_REPLAY_ID_RULE = "boostReplayId"
+    CUSTOM_RULE = "customRule"
 
 
 DEFAULT_BIASES: List[ActivatableBias] = [
@@ -64,6 +65,7 @@ class RuleType(Enum):
     RuleType.BOOST_REPLAY_ID_RULE: 1005,
     RuleType.BOOST_LOW_VOLUME_TRANSACTIONS_RULE: 1400,
     RuleType.BOOST_LATEST_RELEASES_RULE: 1500,
+    RuleType.CUSTOM_RULE: CUSTOM_RULE_START,
 }
 REVERSE_RESERVED_IDS = {value: key for key, value in RESERVED_IDS.items()}
 
diff --git a/src/sentry/migrations/0555_custom_dynamic_sampling_rule.py b/src/sentry/migrations/0555_custom_dynamic_sampling_rule.py
new file mode 100644
index 00000000000000..c74bec336d3c6d
--- /dev/null
+++ b/src/sentry/migrations/0555_custom_dynamic_sampling_rule.py
@@ -0,0 +1,123 @@
+# Generated by Django 3.2.20 on 2023-09-18 11:29
+
+import django.db.models.deletion
+import django.utils.timezone
+from django.db import migrations, models
+
+import sentry.db.models.fields.bounded
+import sentry.db.models.fields.foreignkey
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production. For
+    # the most part, this should only be used for operations where it's safe to run the migration
+    # after your code has deployed. So this should not be used for most operations that alter the
+    # schema of a table.
+    # Here are some things that make sense to mark as dangerous:
+    # - Large data migrations. Typically we want these to be run manually by ops so that they can
+    #   be monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   have ops run this and not block the deploy. Note that while adding an index is a schema
+    #   change, it's completely safe to run the operation after the code has deployed.
+    is_dangerous = False
+
+    dependencies = [
+        ("sentry", "0554_add_team_replica"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="CustomDynamicSamplingRule",
+            fields=[
+                (
+                    "id",
+                    sentry.db.models.fields.bounded.BoundedBigAutoField(
+                        primary_key=True, serialize=False
+                    ),
+                ),
+                ("date_added", models.DateTimeField(default=django.utils.timezone.now)),
+                ("is_active", models.BooleanField(default=True)),
+                ("is_org_level", models.BooleanField(default=False)),
+                ("rule_id", models.IntegerField(default=0)),
+                ("condition", models.TextField()),
+                ("sample_rate", models.FloatField(default=0.0)),
+                ("start_date", models.DateTimeField(default=django.utils.timezone.now)),
+                ("end_date", models.DateTimeField()),
+                ("num_samples", models.IntegerField()),
+                ("condition_hash", models.CharField(max_length=40)),
+                (
+                    "organization",
+                    sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="sentry.organization"
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "sentry_customdynamicsamplingrule",
+            },
+        ),
+        migrations.CreateModel(
+            name="CustomDynamicSamplingRuleProject",
+            fields=[
+                (
+                    "id",
+                    sentry.db.models.fields.bounded.BoundedBigAutoField(
+                        primary_key=True, serialize=False
+                    ),
+                ),
+                (
+                    "custom_dynamic_sampling_rule",
+                    sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="sentry.customdynamicsamplingrule",
+                    ),
+                ),
+                (
+                    "organization",
+                    sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="sentry.organization"
+                    ),
+                ),
+                (
+                    "project",
+                    sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="sentry.project"
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "sentry_customdynamicsamplingruleproject",
+                "unique_together": {("custom_dynamic_sampling_rule", "project")},
+            },
+        ),
+        migrations.AddField(
+            model_name="customdynamicsamplingrule",
+            name="projects",
+            field=models.ManyToManyField(
+                related_name="custom_dynamic_sampling_rules",
+                through="sentry.CustomDynamicSamplingRuleProject",
+                to="sentry.Project",
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="customdynamicsamplingrule",
+            index=models.Index(
+                condition=models.Q(("is_active", True)), fields=["organization"], name="org_idx"
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="customdynamicsamplingrule",
+            index=models.Index(
+                condition=models.Q(("is_active", True)), fields=["end_date"], name="end_date_idx"
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="customdynamicsamplingrule",
+            index=models.Index(
+                condition=models.Q(("is_active", True)),
+                fields=["condition_hash"],
+                name="condition_hash_idx",
+            ),
+        ),
+    ]
diff --git a/src/sentry/models/__init__.py b/src/sentry/models/__init__.py
index 3f77f0c6f88ea9..e69904e8f4c239 100644
--- a/src/sentry/models/__init__.py
+++ b/src/sentry/models/__init__.py
@@ -30,6 +30,7 @@
 from .deletedteam import *  # NOQA
 from .deploy import *  # NOQA
 from .distribution import *  # NOQA
+from .dynamicsampling import *  # NOQA
 from .email import *  # NOQA
 from .environment import *  # NOQA
 from .event import *  # NOQA
diff --git a/src/sentry/models/dynamicsampling.py b/src/sentry/models/dynamicsampling.py
new file mode 100644
index 00000000000000..38f1db6abba8f6
--- /dev/null
+++ b/src/sentry/models/dynamicsampling.py
@@ -0,0 +1,258 @@
+import hashlib
+from datetime import datetime, timedelta
+from typing import Any, Mapping, Optional, Sequence
+
+from django.db import connections, models, router, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from sentry.backup.scopes import RelocationScope
+from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.utils import json
+
+# max number of custom rules that can be created per organization
+MAX_CUSTOM_RULES = 2000
+CUSTOM_RULE_START = 3000
+
+
+class TooManyRules(ValueError):
+    """
+    Raised when a there is already the max number of rules active for an organization
+    """
+
+    pass
+
+
+def get_condition_hash(condition: Any) -> str:
+    """
+    Returns the hash of the rule based on the condition and projects
+    """
+    condition_string = to_order_independent_string(condition)
+    # make it a bit shorter
+    return hashlib.sha1(condition_string.encode("utf-8")).hexdigest()
+
+
+def to_order_independent_string(val: Any) -> str:
+    """
+    Converts a value in an order independent string and then hashes it
+
+    Note: this will insure the same repr is generated for ['x', 'y'] and ['y', 'x']
+        Also the same repr is generated for {'x': 1, 'y': 2} and {'y': 2, 'x': 1}
+    """
+    ret_val = ""
+    if isinstance(val, Mapping):
+        for key in sorted(val.keys()):
+            ret_val += f"{key}:{to_order_independent_string(val[key])}-"
+    elif isinstance(val, (list, tuple)):
+        for item in sorted(val):
+            ret_val += f"{to_order_independent_string(item)}-"
+    else:
+        ret_val = str(val)
+    return ret_val
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRuleProject(Model):
+    """
+    Many-to-many relationship between a custom dynamic sampling rule and a project.
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    custom_dynamic_sampling_rule = FlexibleForeignKey(
+        "sentry.CustomDynamicSamplingRule", on_delete=models.CASCADE
+    )
+    project = FlexibleForeignKey("sentry.Project", on_delete=models.CASCADE)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingruleproject"
+        unique_together = (("custom_dynamic_sampling_rule", "project"),)
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRule(Model):
+    """
+    This represents a custom dynamic sampling rule that is created by the user based
+    on a query (a.k.a. investigation rule).
+
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    date_added = models.DateTimeField(default=timezone.now)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+    projects = models.ManyToManyField(
+        "sentry.Project",
+        related_name="custom_dynamic_sampling_rules",
+        through=CustomDynamicSamplingRuleProject,
+    )
+    is_active = models.BooleanField(default=True)
+    is_org_level = models.BooleanField(default=False)
+    rule_id = models.IntegerField(default=0)
+    condition = models.TextField()
+    sample_rate = models.FloatField(default=0.0)
+    start_date = models.DateTimeField(default=timezone.now)
+    end_date = models.DateTimeField()
+    num_samples = models.IntegerField()
+    condition_hash = models.CharField(max_length=40)
+
+    @property
+    def external_rule_id(self) -> int:
+        """
+        Returns the external rule id
+
+        For external users, i.e. Relay, we need to shift the ids since the slot we
+        have allocated starts at the offset specified in RESERVED_IDS.
+        """
+        return self.rule_id + CUSTOM_RULE_START
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingrule"
+        indexes = [
+            # get active rules for an organization
+            models.Index(fields=["organization"], name="org_idx", condition=Q(is_active=True)),
+            # get expired rules (that are still marked as active)
+            models.Index(fields=["end_date"], name="end_date_idx", condition=Q(is_active=True)),
+            # find active rules for a condition
+            models.Index(
+                fields=["condition_hash"], name="condition_hash_idx", condition=Q(is_active=True)
+            ),
+        ]
+
+    @staticmethod
+    def get_rule_for_org(
+        condition: Any, organization_id: int
+    ) -> Optional["CustomDynamicSamplingRule"]:
+        """
+        Returns an active rule for the given condition and organization if it exists otherwise None
+
+        Note: There should not be more than one active rule for a given condition and organization
+        This function doesn't verify this condition, it just returns the first one.
+        """
+        condition_hash = get_condition_hash(condition)
+        rules = CustomDynamicSamplingRule.objects.filter(
+            organization_id=organization_id,
+            condition_hash=condition_hash,
+            is_active=True,
+            end_date__gt=timezone.now(),
+        )
+        for rule in rules:
+            return rule
+        return None
+
+    @staticmethod
+    def update_or_create(
+        condition: Any,
+        start: datetime,
+        end: datetime,
+        project_ids: Sequence[int],
+        organization_id: int,
+        num_samples: int,
+        sample_rate: float,
+    ) -> "CustomDynamicSamplingRule":
+        with transaction.atomic(router.db_for_write(CustomDynamicSamplingRule)):
+
+            # check if rule already exists for this organization
+            existing_rule = CustomDynamicSamplingRule.get_rule_for_org(condition, organization_id)
+
+            if existing_rule is not None:
+                # we already have an active rule for this condition and this organization
+                # update the expiration date and ensure that our projects are included
+                existing_rule.end_date = max(end, existing_rule.end_date)
+                existing_rule.num_samples = max(num_samples, existing_rule.num_samples)
+                existing_rule.sample_rate = max(sample_rate, existing_rule.sample_rate)
+
+                if not existing_rule.is_org_level:
+                    # for project rules we need to add the projects,org rules already include everything
+                    if len(project_ids) == 0:
+                        # the new rule is an org rule promote current rule to org rule and remove all
+                        # relations to individual projects
+                        existing_rule.is_org_level = True
+                        existing_rule.projects.clear()
+                    else:
+                        # add the new projects to the rule, if not already there
+                        for project_id in project_ids:
+                            CustomDynamicSamplingRuleProject.objects.get_or_create(
+                                custom_dynamic_sampling_rule=existing_rule,
+                                project_id=project_id,
+                                organization_id=organization_id,
+                            )
+                # for org rules we don't need to do anything with the projects
+                existing_rule.save()
+                return existing_rule
+            else:
+                # create a new rule
+                condition_hash = get_condition_hash(condition)
+                is_org_level = len(project_ids) == 0
+                condition_str = json.dumps(condition)
+                rule = CustomDynamicSamplingRule.objects.create(
+                    organization_id=organization_id,
+                    condition=condition_str,
+                    sample_rate=sample_rate,
+                    start_date=start,
+                    end_date=end,
+                    num_samples=num_samples,
+                    condition_hash=condition_hash,
+                    is_active=True,
+                    is_org_level=is_org_level,
+                )
+
+                rule.save()
+                # now try to assign a rule id
+                id = rule.assign_rule_id()
+                if id > MAX_CUSTOM_RULES:
+                    # we have too many rules, delete this one
+                    rule.delete()
+                    raise TooManyRules()
+
+                # set the projects if not org level
+                for project_id in project_ids:
+                    CustomDynamicSamplingRuleProject.objects.get_or_create(
+                        custom_dynamic_sampling_rule=rule,
+                        project_id=project_id,
+                        organization_id=organization_id,
+                    )
+
+                return rule
+
+    def assign_rule_id(self) -> int:
+        """
+        Assigns the smallest rule id that is not taken in the
+        current organization.
+        """
+        table_name = self._meta.db_table
+        if self.id is None:
+            raise ValueError("Cannot assign rule id to unsaved object")
+        if self.rule_id != 0:
+            raise ValueError("Cannot assign rule id to object that already has a rule id")
+
+        now = timezone.now()
+
+        raw_sql = (
+            f"UPDATE {table_name} SET rule_id = ( "
+            f"   SELECT COALESCE ((SELECT MIN(rule_id) + 1  FROM {table_name} WHERE rule_id + 1 NOT IN ("
+            f"       SELECT rule_id FROM {table_name} WHERE organization_id = %s AND end_date > %s AND "
+            f"is_active)),1))  "
+            f"WHERE id = %s"
+        )
+        with connections["default"].cursor() as cursor:
+            cursor.execute(raw_sql, (self.organization.id, now, self.id))
+        self.refresh_from_db()
+        return self.rule_id
+
+    @staticmethod
+    def deactivate_old_rules() -> None:
+        """
+        Deactivates all rules expired rules (this is just an optimization to remove old rules from indexes).
+
+        This should be called periodically to clean up old rules (it is not necessary to call it for correctness,
+        just for performance)
+        """
+        CustomDynamicSamplingRule.objects.filter(
+            # give it a minute grace period to make sure we don't deactivate rules that are still active
+            end_date__lt=timezone.now()
+            - timedelta(minutes=1),
+        ).update(is_active=False)
diff --git a/tests/sentry/api/endpoints/test_custom_rules.py b/tests/sentry/api/endpoints/test_custom_rules.py
new file mode 100644
index 00000000000000..9f7f0974e86e5a
--- /dev/null
+++ b/tests/sentry/api/endpoints/test_custom_rules.py
@@ -0,0 +1,180 @@
+from datetime import datetime, timedelta
+
+import pytest
+
+from sentry.api.endpoints.custom_rules import (
+    _DATE_FORMAT,
+    DEFAULT_PERIOD_STRING,
+    MAX_RULE_PERIOD_STRING,
+    CustomRulesInputSerializer,
+)
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.helpers import Feature
+from sentry.testutils.silo import region_silo_test
+
+
+@region_silo_test(stable=True)
+class CustomRulesEndpoint(APITestCase):
+    """
+    Tests that calling the endpoint converts the query to a rule returns it and saves it in the db
+    """
+
+    endpoint = "sentry-api-0-organization-dynamic_sampling-custom_rules"
+    method = "post"
+
+    def setUp(self):
+        super().setUp()
+        self.login_as(user=self.user)
+        self.second_project = self.create_project(organization=self.organization)
+
+    def test_simple(self):
+        request_data = {
+            "query": "event.type:transaction",
+            "projects": [self.project.id],
+            "period": "1h",
+            "overrideExisting": True,
+        }
+        with Feature({"organizations:investigation-bias": True}):
+            resp = self.get_response(self.organization.slug, raw_data=request_data)
+
+        assert resp.status_code == 200
+
+        data = resp.data
+
+        start_date = datetime.strptime(data["startDate"], _DATE_FORMAT)
+        end_date = datetime.strptime(data["endDate"], _DATE_FORMAT)
+        assert end_date - start_date == timedelta(hours=1)
+        projects = data["projects"]
+        assert projects == [self.project.id]
+        org_id = data["orgId"]
+        assert org_id == self.organization.id
+
+        # check the database
+        rule_id = data["ruleId"]
+        rules = list(self.organization.customdynamicsamplingrule_set.all())
+        assert len(rules) == 1
+        rule = rules[0]
+        assert rule.external_rule_id == rule_id
+
+    def test_updates_existing(self):
+        """
+        Test that the endpoint updates an existing rule if the same rule condition is given
+
+        The rule id should be the same
+        The period and the projects should be updated
+        """
+        request_data = {
+            "query": "event.type:transaction",
+            "projects": [self.project.id],
+            "period": "1h",
+            "overrideExisting": True,
+        }
+
+        # create rule
+        with Feature({"organizations:investigation-bias": True}):
+            resp = self.get_response(self.organization.slug, raw_data=request_data)
+
+        assert resp.status_code == 200
+
+        data = resp.data
+
+        rule_id = data["ruleId"]
+        start_date = datetime.strptime(data["startDate"], _DATE_FORMAT)
+        end_date = datetime.strptime(data["endDate"], _DATE_FORMAT)
+        assert end_date - start_date == timedelta(hours=1)
+
+        request_data = {
+            "query": "event.type:transaction",
+            "projects": [self.second_project.id],
+            "period": "2h",
+            "overrideExisting": True,
+        }
+
+        # update existing rule
+        with Feature({"organizations:investigation-bias": True}):
+            resp = self.get_response(self.organization.slug, raw_data=request_data)
+
+        assert resp.status_code == 200
+        data = resp.data
+
+        start_date = datetime.strptime(data["startDate"], _DATE_FORMAT)
+        end_date = datetime.strptime(data["endDate"], _DATE_FORMAT)
+        assert end_date - start_date >= timedelta(hours=2)
+
+        projects = data["projects"]
+        assert self.project.id in projects
+        assert self.second_project.id in projects
+
+        new_rule_id = data["ruleId"]
+        assert rule_id == new_rule_id
+
+    def test_checks_feature(self):
+        """
+        Checks request fails without the feature
+        """
+        request_data = {
+            "query": "event.type:transaction",
+            "projects": [self.project.id],
+            "period": "1h",
+            "overrideExisting": True,
+        }
+        with Feature({"organizations:investigation-bias": False}):
+            resp = self.get_response(self.organization.slug, raw_data=request_data)
+
+        assert resp.status_code == 404
+
+
+@pytest.mark.parametrize(
+    "what,value,valid",
+    [
+        ("query", "event.type:transaction", True),
+        ("projects", [1, 2, 3], True),
+        ("period", "1h", True),
+        ("projects", ["abc"], False),
+        ("period", "hello", False),
+    ],
+)
+def test_custom_rule_serializer(what, value, valid):
+    """
+    Test that the serializer works as expected
+    """
+    data = {"query": "event.type:transaction", "projects": [1, 2, 3], "period": "1h"}
+    data[what] = value
+
+    serializer = CustomRulesInputSerializer(data=data)
+
+    assert serializer.is_valid() == valid
+
+
+def test_custom_rule_serializer_default_period():
+    """
+    Test that the serializer validation sets the default period
+    """
+    data = {"query": "event.type:transaction", "projects": [1, 2, 3]}
+    serializer = CustomRulesInputSerializer(data=data)
+
+    assert serializer.is_valid()
+    assert serializer.validated_data["period"] == DEFAULT_PERIOD_STRING
+
+
+def test_custom_rule_serializer_limits_period():
+    """
+    Test that the serializer validation limits the peroid to the max allowed
+    """
+    data = {"query": "event.type:transaction", "projects": [1, 2, 3], "period": "100d"}
+    serializer = CustomRulesInputSerializer(data=data)
+
+    assert serializer.is_valid()
+    assert serializer.validated_data["period"] == MAX_RULE_PERIOD_STRING
+
+
+def test_custom_rule_serializer_creates_org_rule_when_no_projects_given():
+    """
+    Test that the serializer creates an org level rule when no projects are given
+    """
+    data = {"query": "event.type:transaction", "period": "1h"}
+    serializer = CustomRulesInputSerializer(data=data)
+
+    assert serializer.is_valid()
+    # an org level rule has an empty list of projects set
+    assert serializer.validated_data["projects"] == []
diff --git a/tests/sentry/backup/test_models.py b/tests/sentry/backup/test_models.py
index d8a382d2cafab9..d0d34b023505af 100644
--- a/tests/sentry/backup/test_models.py
+++ b/tests/sentry/backup/test_models.py
@@ -24,6 +24,7 @@
     PendingIncidentSnapshot,
     TimeSeriesSnapshot,
 )
+from sentry.models import CustomDynamicSamplingRule, CustomDynamicSamplingRuleProject
 from sentry.models.actor import Actor
 from sentry.models.apiapplication import ApiApplication
 from sentry.models.apiauthorization import ApiAuthorization
@@ -525,6 +526,25 @@ def test_user_role(self):
         UserRoleUser.objects.create(user=user, role=role)
         return self.import_export_then_validate()
 
+    @targets(mark(CustomDynamicSamplingRule, CustomDynamicSamplingRuleProject))
+    def test_custom_dynamic_sampling_rule(self):
+        condition = {"op": "equals", "name": "environment", "value": "prod"}
+
+        rule = CustomDynamicSamplingRule.update_or_create(
+            condition=condition,
+            start=timezone.now(),
+            end=timezone.now() + timedelta(hours=1),
+            project_ids=[self.project.id],
+            organization_id=self.organization.id,
+            num_samples=100,
+            sample_rate=0.5,
+        )
+
+        CustomDynamicSamplingRuleProject.objects.create(
+            rule=rule, project=self.project, organization=self.organization
+        )
+        return self.import_export_then_validate()
+
 
 @run_backup_tests_only_on_single_db
 class DynamicRelocationScopeTests(TransactionTestCase):
diff --git a/tests/sentry/models/test_dynamic_sampling.py b/tests/sentry/models/test_dynamic_sampling.py
new file mode 100644
index 00000000000000..1b6e692bd5922f
--- /dev/null
+++ b/tests/sentry/models/test_dynamic_sampling.py
@@ -0,0 +1,139 @@
+from datetime import timedelta
+
+from django.utils import timezone
+from freezegun import freeze_time
+
+from sentry.models import CustomDynamicSamplingRule
+from sentry.testutils.cases import TestCase
+from sentry.testutils.silo import region_silo_test
+
+
+@freeze_time("2023-09-18")
+@region_silo_test()
+class TestCustomDynamicSamplingRuleProject(TestCase):
+    def setUp(self):
+        super().setUp()
+        self.second_project = self.create_project()
+
+    def test_update_or_create(self):
+        condition = {"op": "equals", "name": "environment", "value": "prod"}
+
+        end1 = timezone.now() + timedelta(hours=1)
+
+        rule = CustomDynamicSamplingRule.update_or_create(
+            condition=condition,
+            start=timezone.now(),
+            end=end1,
+            project_ids=[self.project.id],
+            organization_id=self.organization.id,
+            num_samples=100,
+            sample_rate=0.5,
+        )
+
+        end2 = timezone.now() + timedelta(hours=1)
+        updated_rule = CustomDynamicSamplingRule.update_or_create(
+            condition=condition,
+            start=timezone.now() + timedelta(minutes=1),
+            end=end2,
+            project_ids=[self.second_project.id],
+            organization_id=self.organization.id,
+            num_samples=100,
+            sample_rate=0.5,
+        )
+
+        assert rule.id == updated_rule.id
+        projects = updated_rule.projects.all()
+
+        assert len(projects) == 2
+        assert self.project in projects
+        assert self.second_project in projects
+
+        assert updated_rule.end_date >= end1
+        assert updated_rule.end_date >= end2
+
+    def test_assign_rule_id(self):
+        def create_rule_for_env(env_idx: int):
+            condition = {"op": "equals", "name": "environment", "value": f"prod{env_idx}"}
+            return CustomDynamicSamplingRule.update_or_create(
+                condition=condition,
+                start=timezone.now(),
+                end=timezone.now() + timedelta(hours=1),
+                project_ids=[self.project.id],
+                organization_id=self.organization.id,
+                num_samples=100,
+                sample_rate=0.5,
+            )
+
+        rule_ids = set()
+        rules = []
+        for idx in range(3):
+            rule = create_rule_for_env(idx)
+            rule_ids.add(rule.rule_id)
+            rules.append(rule)
+
+        # all 3 rules have different rule ids
+        assert len(rule_ids) == 3
+
+        # make a rule obsolete and check that the rule id is reused
+        rules[1].is_active = False
+        rules[1].save()
+
+        new_rule = create_rule_for_env(4)
+        assert new_rule.rule_id == rules[1].rule_id
+
+        # a new rule will take another slot (now that there is no free slot)
+        new_rule_2 = create_rule_for_env(5)
+        assert new_rule_2.rule_id not in rule_ids
+
+        # make again an empty slot ( this time by having the rule expire)
+        rules[2].start_date = timezone.now() - timedelta(hours=2)
+        rules[2].end_date = timezone.now() - timedelta(hours=1)
+        rules[2].save()
+
+        # the new rule should take the empty slot
+        new_rule_3 = create_rule_for_env(6)
+        assert new_rule_3.rule_id == rules[2].rule_id
+
+    def test_deactivate_old_rules(self):
+        idx = 1
+
+        old_rules = []
+        new_rules = []
+
+        def create_rule(is_old: bool, idx: int):
+            condition = {"op": "equals", "name": "environment", "value": f"prod{idx}"}
+            if is_old:
+                end_delta = -timedelta(hours=1)
+            else:
+                end_delta = timedelta(hours=1)
+            return CustomDynamicSamplingRule.update_or_create(
+                condition=condition,
+                start=timezone.now() - timedelta(hours=2),
+                end=timezone.now() + end_delta,
+                project_ids=[self.project.id],
+                organization_id=self.organization.id,
+                num_samples=100,
+                sample_rate=0.5,
+            )
+
+        for i in range(10):
+            for is_old in [True, False]:
+                idx += 1
+                rule = create_rule(is_old, idx)
+                if is_old:
+                    old_rules.append(rule)
+                else:
+                    new_rules.append(rule)
+
+        CustomDynamicSamplingRule.deactivate_old_rules()
+
+        # check that all old rules are inactive and all new rules are active
+        inactive_rules = list(CustomDynamicSamplingRule.objects.filter(is_active=False))
+        assert len(inactive_rules) == 10
+        for rule in old_rules:
+            assert rule in inactive_rules
+
+        active_rules = list(CustomDynamicSamplingRule.objects.filter(is_active=True))
+        assert len(active_rules) == 10
+        for rule in new_rules:
+            assert rule in active_rules


Code review comments:
diff_hunk: @@ -0,0 +1,258 @@
+import hashlib
+from datetime import datetime, timedelta
+from typing import Any, Mapping, Optional, Sequence
+
+from django.db import connections, models, router, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from sentry.backup.scopes import RelocationScope
+from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.utils import json
+
+# max number of custom rules that can be created per organization
+MAX_CUSTOM_RULES = 2000
+CUSTOM_RULE_START = 3000
+
+
+class TooManyRules(ValueError):
+    """
+    Raised when a there is already the max number of rules active for an organization
+    """
+
+    pass
+
+
+def get_condition_hash(condition: Any) -> str:
+    """
+    Returns the hash of the rule based on the condition and projects
+    """
+    condition_string = to_order_independent_string(condition)
+    # make it a bit shorter
+    return hashlib.sha1(condition_string.encode("utf-8")).hexdigest()
+
+
+def to_order_independent_string(val: Any) -> str:
+    """
+    Converts a value in an order independent string and then hashes it
+
+    Note: this will insure the same repr is generated for ['x', 'y'] and ['y', 'x']
+        Also the same repr is generated for {'x': 1, 'y': 2} and {'y': 2, 'x': 1}
+    """
+    ret_val = ""
+    if isinstance(val, Mapping):
+        for key in sorted(val.keys()):
+            ret_val += f"{key}:{to_order_independent_string(val[key])}-"
+    elif isinstance(val, (list, tuple)):
+        for item in sorted(val):
+            ret_val += f"{to_order_independent_string(item)}-"
+    else:
+        ret_val = str(val)
+    return ret_val
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRuleProject(Model):
+    """
+    Many-to-many relationship between a custom dynamic sampling rule and a project.
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    custom_dynamic_sampling_rule = FlexibleForeignKey(
+        "sentry.CustomDynamicSamplingRule", on_delete=models.CASCADE
+    )
+    project = FlexibleForeignKey("sentry.Project", on_delete=models.CASCADE)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
path: src/sentry/models/dynamicsampling.py
text: @user1:
Why do you have a foreign key to the organization table here?

@user1:
Can't you directly infer it from the main table?

@author:
It was suggested by @user2-auer to simplify cleanup jobs.
It is not super useful ( maybe we should remove it) , if we do delete it, it also simplifies the code ( since I don't need to define the through model) and I can associate projects with rules easier (just by adding the project to the projects set) no need to explicitly create the entry in the through table. 

@user2-auer What do you think ?

diff_hunk: @@ -0,0 +1,258 @@
+import hashlib
+from datetime import datetime, timedelta
+from typing import Any, Mapping, Optional, Sequence
+
+from django.db import connections, models, router, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from sentry.backup.scopes import RelocationScope
+from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.utils import json
+
+# max number of custom rules that can be created per organization
+MAX_CUSTOM_RULES = 2000
+CUSTOM_RULE_START = 3000
+
+
+class TooManyRules(ValueError):
+    """
+    Raised when a there is already the max number of rules active for an organization
+    """
+
+    pass
+
+
+def get_condition_hash(condition: Any) -> str:
+    """
+    Returns the hash of the rule based on the condition and projects
+    """
+    condition_string = to_order_independent_string(condition)
+    # make it a bit shorter
+    return hashlib.sha1(condition_string.encode("utf-8")).hexdigest()
+
+
+def to_order_independent_string(val: Any) -> str:
+    """
+    Converts a value in an order independent string and then hashes it
+
+    Note: this will insure the same repr is generated for ['x', 'y'] and ['y', 'x']
+        Also the same repr is generated for {'x': 1, 'y': 2} and {'y': 2, 'x': 1}
+    """
+    ret_val = ""
+    if isinstance(val, Mapping):
+        for key in sorted(val.keys()):
+            ret_val += f"{key}:{to_order_independent_string(val[key])}-"
+    elif isinstance(val, (list, tuple)):
+        for item in sorted(val):
+            ret_val += f"{to_order_independent_string(item)}-"
+    else:
+        ret_val = str(val)
+    return ret_val
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRuleProject(Model):
+    """
+    Many-to-many relationship between a custom dynamic sampling rule and a project.
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    custom_dynamic_sampling_rule = FlexibleForeignKey(
+        "sentry.CustomDynamicSamplingRule", on_delete=models.CASCADE
+    )
+    project = FlexibleForeignKey("sentry.Project", on_delete=models.CASCADE)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingruleproject"
+        unique_together = (("custom_dynamic_sampling_rule", "project"),)
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRule(Model):
+    """
+    This represents a custom dynamic sampling rule that is created by the user based
+    on a query (a.k.a. investigation rule).
+
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    date_added = models.DateTimeField(default=timezone.now)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+    projects = models.ManyToManyField(
+        "sentry.Project",
+        related_name="custom_dynamic_sampling_rules",
+        through=CustomDynamicSamplingRuleProject,
+    )
+    is_active = models.BooleanField(default=True)
+    is_org_level = models.BooleanField(default=False)
+    rule_id = models.IntegerField(default=0)
+    condition = models.TextField()
+    sample_rate = models.FloatField(default=0.0)
+    start_date = models.DateTimeField(default=timezone.now)
+    end_date = models.DateTimeField()
+    num_samples = models.IntegerField()
+    condition_hash = models.CharField(max_length=40)
+
+    @property
+    def external_rule_id(self) -> int:
+        """
+        Returns the external rule id
+
+        For external users, i.e. Relay, we need to shift the ids since the slot we
+        have allocated starts at the offset specified in RESERVED_IDS.
+        """
+        return self.rule_id + CUSTOM_RULE_START
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingrule"
+        indexes = [
+            # get active rules for an organization
+            models.Index(fields=["organization"], name="org_idx", condition=Q(is_active=True)),
+            # get expired rules (that are still marked as active)
+            models.Index(fields=["end_date"], name="end_date_idx", condition=Q(is_active=True)),
+            # find active rules for a condition
+            models.Index(
+                fields=["condition_hash"], name="condition_hash_idx", condition=Q(is_active=True)
+            ),
+        ]
+
+    @staticmethod
+    def get_rule_for_org(
+        condition: Any, organization_id: int
+    ) -> Optional["CustomDynamicSamplingRule"]:
+        """
+        Returns an active rule for the given condition and organization if it exists otherwise None
+
+        Note: There should not be more than one active rule for a given condition and organization
+        This function doesn't verify this condition, it just returns the first one.
+        """
+        condition_hash = get_condition_hash(condition)
+        rules = CustomDynamicSamplingRule.objects.filter(
+            organization_id=organization_id,
+            condition_hash=condition_hash,
+            is_active=True,
+            end_date__gt=timezone.now(),
+        )
+        for rule in rules:
+            return rule
+        return None
path: src/sentry/models/dynamicsampling.py
text: @user1:
```suggestion
        return rules[0] if rules else None
```

@author:
Not sure how the QuerySet responds to this ( i.e. `if rules:`), if it was a list it would make sense, 
it may work with QuerySet as well but I didn't try it.

@author:
Checked, it works, changed

diff_hunk: @@ -0,0 +1,258 @@
+import hashlib
+from datetime import datetime, timedelta
+from typing import Any, Mapping, Optional, Sequence
+
+from django.db import connections, models, router, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from sentry.backup.scopes import RelocationScope
+from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.utils import json
+
+# max number of custom rules that can be created per organization
+MAX_CUSTOM_RULES = 2000
+CUSTOM_RULE_START = 3000
+
+
+class TooManyRules(ValueError):
+    """
+    Raised when a there is already the max number of rules active for an organization
+    """
+
+    pass
+
+
+def get_condition_hash(condition: Any) -> str:
+    """
+    Returns the hash of the rule based on the condition and projects
+    """
+    condition_string = to_order_independent_string(condition)
+    # make it a bit shorter
+    return hashlib.sha1(condition_string.encode("utf-8")).hexdigest()
+
+
+def to_order_independent_string(val: Any) -> str:
+    """
+    Converts a value in an order independent string and then hashes it
+
+    Note: this will insure the same repr is generated for ['x', 'y'] and ['y', 'x']
+        Also the same repr is generated for {'x': 1, 'y': 2} and {'y': 2, 'x': 1}
+    """
+    ret_val = ""
+    if isinstance(val, Mapping):
+        for key in sorted(val.keys()):
+            ret_val += f"{key}:{to_order_independent_string(val[key])}-"
+    elif isinstance(val, (list, tuple)):
+        for item in sorted(val):
+            ret_val += f"{to_order_independent_string(item)}-"
+    else:
+        ret_val = str(val)
+    return ret_val
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRuleProject(Model):
+    """
+    Many-to-many relationship between a custom dynamic sampling rule and a project.
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    custom_dynamic_sampling_rule = FlexibleForeignKey(
+        "sentry.CustomDynamicSamplingRule", on_delete=models.CASCADE
+    )
+    project = FlexibleForeignKey("sentry.Project", on_delete=models.CASCADE)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingruleproject"
+        unique_together = (("custom_dynamic_sampling_rule", "project"),)
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRule(Model):
+    """
+    This represents a custom dynamic sampling rule that is created by the user based
+    on a query (a.k.a. investigation rule).
+
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    date_added = models.DateTimeField(default=timezone.now)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+    projects = models.ManyToManyField(
+        "sentry.Project",
+        related_name="custom_dynamic_sampling_rules",
+        through=CustomDynamicSamplingRuleProject,
+    )
+    is_active = models.BooleanField(default=True)
+    is_org_level = models.BooleanField(default=False)
+    rule_id = models.IntegerField(default=0)
+    condition = models.TextField()
+    sample_rate = models.FloatField(default=0.0)
+    start_date = models.DateTimeField(default=timezone.now)
+    end_date = models.DateTimeField()
+    num_samples = models.IntegerField()
+    condition_hash = models.CharField(max_length=40)
+
+    @property
+    def external_rule_id(self) -> int:
+        """
+        Returns the external rule id
+
+        For external users, i.e. Relay, we need to shift the ids since the slot we
+        have allocated starts at the offset specified in RESERVED_IDS.
+        """
+        return self.rule_id + CUSTOM_RULE_START
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingrule"
+        indexes = [
+            # get active rules for an organization
+            models.Index(fields=["organization"], name="org_idx", condition=Q(is_active=True)),
+            # get expired rules (that are still marked as active)
+            models.Index(fields=["end_date"], name="end_date_idx", condition=Q(is_active=True)),
+            # find active rules for a condition
+            models.Index(
+                fields=["condition_hash"], name="condition_hash_idx", condition=Q(is_active=True)
+            ),
+        ]
+
+    @staticmethod
+    def get_rule_for_org(
+        condition: Any, organization_id: int
+    ) -> Optional["CustomDynamicSamplingRule"]:
+        """
+        Returns an active rule for the given condition and organization if it exists otherwise None
+
+        Note: There should not be more than one active rule for a given condition and organization
+        This function doesn't verify this condition, it just returns the first one.
+        """
+        condition_hash = get_condition_hash(condition)
+        rules = CustomDynamicSamplingRule.objects.filter(
+            organization_id=organization_id,
+            condition_hash=condition_hash,
+            is_active=True,
+            end_date__gt=timezone.now(),
+        )
+        for rule in rules:
+            return rule
+        return None
+
+    @staticmethod
+    def update_or_create(
+        condition: Any,
+        start: datetime,
+        end: datetime,
+        project_ids: Sequence[int],
+        organization_id: int,
+        num_samples: int,
+        sample_rate: float,
+    ) -> "CustomDynamicSamplingRule":
+        with transaction.atomic(router.db_for_write(CustomDynamicSamplingRule)):
+
path: src/sentry/models/dynamicsampling.py
text: @user1:
```suggestion

```

diff_hunk: @@ -0,0 +1,258 @@
+import hashlib
+from datetime import datetime, timedelta
+from typing import Any, Mapping, Optional, Sequence
+
+from django.db import connections, models, router, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from sentry.backup.scopes import RelocationScope
+from sentry.db.models import FlexibleForeignKey, Model, region_silo_only_model
+from sentry.utils import json
+
+# max number of custom rules that can be created per organization
+MAX_CUSTOM_RULES = 2000
+CUSTOM_RULE_START = 3000
+
+
+class TooManyRules(ValueError):
+    """
+    Raised when a there is already the max number of rules active for an organization
+    """
+
+    pass
+
+
+def get_condition_hash(condition: Any) -> str:
+    """
+    Returns the hash of the rule based on the condition and projects
+    """
+    condition_string = to_order_independent_string(condition)
+    # make it a bit shorter
+    return hashlib.sha1(condition_string.encode("utf-8")).hexdigest()
+
+
+def to_order_independent_string(val: Any) -> str:
+    """
+    Converts a value in an order independent string and then hashes it
+
+    Note: this will insure the same repr is generated for ['x', 'y'] and ['y', 'x']
+        Also the same repr is generated for {'x': 1, 'y': 2} and {'y': 2, 'x': 1}
+    """
+    ret_val = ""
+    if isinstance(val, Mapping):
+        for key in sorted(val.keys()):
+            ret_val += f"{key}:{to_order_independent_string(val[key])}-"
+    elif isinstance(val, (list, tuple)):
+        for item in sorted(val):
+            ret_val += f"{to_order_independent_string(item)}-"
+    else:
+        ret_val = str(val)
+    return ret_val
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRuleProject(Model):
+    """
+    Many-to-many relationship between a custom dynamic sampling rule and a project.
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    custom_dynamic_sampling_rule = FlexibleForeignKey(
+        "sentry.CustomDynamicSamplingRule", on_delete=models.CASCADE
+    )
+    project = FlexibleForeignKey("sentry.Project", on_delete=models.CASCADE)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingruleproject"
+        unique_together = (("custom_dynamic_sampling_rule", "project"),)
+
+
+@region_silo_only_model
+class CustomDynamicSamplingRule(Model):
+    """
+    This represents a custom dynamic sampling rule that is created by the user based
+    on a query (a.k.a. investigation rule).
+
+    """
+
+    __relocation_scope__ = RelocationScope.Organization
+
+    date_added = models.DateTimeField(default=timezone.now)
+    organization = FlexibleForeignKey("sentry.Organization", on_delete=models.CASCADE)
+    projects = models.ManyToManyField(
+        "sentry.Project",
+        related_name="custom_dynamic_sampling_rules",
+        through=CustomDynamicSamplingRuleProject,
+    )
+    is_active = models.BooleanField(default=True)
+    is_org_level = models.BooleanField(default=False)
+    rule_id = models.IntegerField(default=0)
+    condition = models.TextField()
+    sample_rate = models.FloatField(default=0.0)
+    start_date = models.DateTimeField(default=timezone.now)
+    end_date = models.DateTimeField()
+    num_samples = models.IntegerField()
+    condition_hash = models.CharField(max_length=40)
+
+    @property
+    def external_rule_id(self) -> int:
+        """
+        Returns the external rule id
+
+        For external users, i.e. Relay, we need to shift the ids since the slot we
+        have allocated starts at the offset specified in RESERVED_IDS.
+        """
+        return self.rule_id + CUSTOM_RULE_START
+
+    class Meta:
+        app_label = "sentry"
+        db_table = "sentry_customdynamicsamplingrule"
+        indexes = [
+            # get active rules for an organization
+            models.Index(fields=["organization"], name="org_idx", condition=Q(is_active=True)),
+            # get expired rules (that are still marked as active)
+            models.Index(fields=["end_date"], name="end_date_idx", condition=Q(is_active=True)),
+            # find active rules for a condition
+            models.Index(
+                fields=["condition_hash"], name="condition_hash_idx", condition=Q(is_active=True)
+            ),
+        ]
+
+    @staticmethod
+    def get_rule_for_org(
+        condition: Any, organization_id: int
+    ) -> Optional["CustomDynamicSamplingRule"]:
+        """
+        Returns an active rule for the given condition and organization if it exists otherwise None
+
+        Note: There should not be more than one active rule for a given condition and organization
+        This function doesn't verify this condition, it just returns the first one.
+        """
+        condition_hash = get_condition_hash(condition)
+        rules = CustomDynamicSamplingRule.objects.filter(
+            organization_id=organization_id,
+            condition_hash=condition_hash,
+            is_active=True,
+            end_date__gt=timezone.now(),
+        )
path: src/sentry/models/dynamicsampling.py
text: @user1:
```suggestion
        )[:1]
```

With this you will optimize the query by using `LIMIT`. Of course you will have to check if the result is empty and in that case you will return `None`.

@author:
the query can't return more than one (no need for limit), I don't create multiple rules with the same condition in an org  ( maybe I should create a uniq index on [condition_hash, org_id], but didn't want to complicate things).

Anyway the [:1] is a good suggestion