PSAv2 endpoint improvements (#36782)

* Updates search opts and cursor for searching property fields

* Adds store changes and tests to accomodate the new search options

* Add an index for improving delta query efficiency

* Enhances the get property fields endpoint to support deltas and hierarchical searches

* Adds new fields search endpoint

* Adds since to the property values endpoints

* Adds property value endpoint documentation

* Fix linter

* Address coderabbit comments

* Complete i18n

* Minor improvements

* Consistently apply DWIM semantics around object_type system

* Add support for DM/GMs

* Move auditRec to the top of the searchPropertyFieldsCore method

* Update since mechanics to use greater or equal

* Update API docs

* Updated tests to filter after the semantics changes

---------

Co-authored-by: Miguel de la Cruz <miguel@ctrlz.es>

api/v4/source/properties.yaml

@@ -99,7 +99,31 @@
         - properties
       summary: Get property fields
       description: >
-        Get a list of property fields for a specific group and object type. Requires a target_type parameter to scope the query, except when `object_type=system` — in that case `target_type` is implicit and any value supplied is ignored. Filter further by target_id to narrow results. Uses cursor-based pagination.
+        Get a list of property fields for a specific group and object type. Uses cursor-based pagination.
+
+
+        **Scope modes (mutually exclusive):**
+
+
+        - **Hierarchical scope** (`channel_id` and/or `team_id`): returns fields scoped to the named resource *and every ancestor scope above it*. System-level rows are always included; team-level rows are included when `team_id` is set; channel-level rows are included only when `channel_id` is set. When `channel_id` is provided, the channel's team is resolved server-side, so any `team_id` the caller passes is ignored. For DM and GM channels there is no parent team, so the hierarchy collapses to `system → channel` and no team-scoped rows are returned. Requires the caller to have `read_channel` for the channel and/or `view_team` for the team.
+
+
+        - **Single-target scope** (`target_type` + `target_id`): returns fields scoped to exactly one resource. `target_type` must be `system`, `team`, or `channel`; `target_id` is required for `team` and `channel`.
+
+
+        Mixing the two modes (any of `channel_id`/`team_id` together with any of `target_type`/`target_id`) returns 400. Omitting both modes also returns 400, except for the system case below.
+
+
+        **System-object:** when `object_type` is `system`, the endpoint always resolves to `target_type=system` and ignores any `channel_id`, `team_id`, `target_type`, or `target_id` the caller may have passed. System-object fields can only live at the system scope by invariant, so any other scope is a semantic no-op rather than an error.
+
+
+        **Delta sync via `since`:** When `since > 0`, the endpoint returns rows whose `update_at` is **greater than or equal to** the cutoff, *including* tombstoned rows (`delete_at > 0`). The inclusive boundary lets clients safely re-use the highest `update_at` seen on a previous page as the next `since` value without missing rows that happened to share that millisecond. Rows within a delta page are ordered by `(update_at, id)` and the cursor disambiguates same-millisecond rows across pages.
+
+
+        **Cursor source:** the client owns the cursor. The initial delta-sync request omits `since` (or sends `since=0`). For subsequent requests the client persists the `update_at` and `id` of the last row in the response and passes them back as `cursor_update_at` and `cursor_id` alongside the same `since` value. This mirrors the cursor convention used by the shared-channel post-sync API.
+
+
+        **Cursor key must match the active ordering:** in delta mode (`since > 0`) the endpoint orders by `update_at` and pagination requires `cursor_update_at`; otherwise it orders by `create_at` and pagination requires `cursor_create_at`. Passing the wrong key for the active mode returns 400.
       operationId: GetPropertyFields
       parameters:
         - name: group_name
@@ -114,10 +138,32 @@
           required: true
           schema:
             type: string
+        - name: channel_id
+          in: query
+          description: >
+            Hierarchical scope. When set, the response includes system-level
+            rows, team-level rows for the channel's team, and channel-level
+            rows for this channel. Mutually exclusive with `target_type` and
+            `target_id`. Requires `read_channel` on the channel.
+          schema:
+            type: string
+        - name: team_id
+          in: query
+          description: >
+            Hierarchical scope. When set without `channel_id`, the response
+            includes system-level and team-level rows for this team. When
+            `channel_id` is also set, this value is ignored — the channel's
+            team is resolved server-side and used instead. Mutually exclusive
+            with `target_type` and `target_id`. Requires `view_team` on the
+            team.
+          schema:
+            type: string
         - name: target_type
           in: query
-          description: The scope level to query. Must be one of 'system', 'team', or 'channel'.
-          required: true
+          description: >
+            Single-target scope. One of `system`, `team`, or `channel`.
+            Required in single-target mode. Mutually exclusive with `channel_id`
+            and `team_id`.
           schema:
             type: string
             enum:
@@ -126,9 +172,20 @@
               - channel
         - name: target_id
           in: query
-          description: Filter by target ID. Required when target_type is 'channel' or 'team'.
+          description: >
+            Single-target scope. Required when `target_type` is `channel` or
+            `team`. Mutually exclusive with `channel_id` and `team_id`.
           schema:
             type: string
+        - name: since
+          in: query
+          description: >
+            Unix timestamp in milliseconds. When greater than 0, returns
+            fields with `update_at` greater than or equal to this value,
+            including tombstones.
+          schema:
+            type: integer
+            format: int64
         - name: cursor_id
           in: query
           description: The ID of the last property field from the previous page, for cursor-based pagination.
@@ -136,7 +193,19 @@
             type: string
         - name: cursor_create_at
           in: query
-          description: The create_at timestamp of the last property field from the previous page. Must be provided together with cursor_id.
+          description: >
+            The `create_at` timestamp of the last property field from the
+            previous page. Required alongside `cursor_id` when `since` is
+            absent. Mutually exclusive with `cursor_update_at`.
+          schema:
+            type: integer
+            format: int64
+        - name: cursor_update_at
+          in: query
+          description: >
+            The `update_at` timestamp of the last property field from the
+            previous page. Required alongside `cursor_id` when `since` is
+            present. Mutually exclusive with `cursor_create_at`.
           schema:
             type: integer
             format: int64
@@ -161,6 +230,142 @@
           $ref: "#/components/responses/Unauthorized"
         "403":
           $ref: "#/components/responses/Forbidden"
+        "404":
+          $ref: "#/components/responses/NotFound"
+  "/api/v4/properties/groups/{group_name}/fields/search":
+    post:
+      tags:
+        - properties
+      summary: Search property fields across multiple object types
+      description: >
+        Returns matching fields across every requested object type in one
+        response. The request body is a `PropertyFieldSearch` object whose
+        `object_types` field lists the object types to include.
+
+        Scope, `since`, cursor, and permission semantics are identical to the
+        get property fields endpoint, including the system-object collapse:
+        when `object_types` is exactly `["system"]`, any scope or target
+        params in the body are ignored and the endpoint resolves to
+        `target_type=system`. Any other combination without an explicit
+        scope returns 400.
+
+        Requesting a single value in `object_types` is equivalent to calling
+        the singular endpoint and is supported for client uniformity.
+      operationId: SearchPropertyFields
+      parameters:
+        - name: group_name
+          in: path
+          description: The name of the property group
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required:
+                - object_types
+              properties:
+                object_types:
+                  type: array
+                  minItems: 1
+                  description: >
+                    One or more object types to include in the response. At
+                    least one value is required; unknown values return 400.
+                  items:
+                    type: string
+                    enum:
+                      - post
+                      - channel
+                      - user
+                      - template
+                      - system
+                channel_id:
+                  type: string
+                  description: >
+                    Hierarchical scope. When set, the response includes
+                    system-level rows, team-level rows for the channel's
+                    team, and channel-level rows for this channel across
+                    every requested `object_types`. Mutually exclusive with
+                    `target_type`/`target_id`. Requires `read_channel` on
+                    the channel.
+                team_id:
+                  type: string
+                  description: >
+                    Hierarchical scope. When set without `channel_id`, the
+                    response includes system-level and team-level rows for
+                    this team across every requested `object_types`. When
+                    `channel_id` is also set, this value is ignored — the
+                    channel's team is resolved server-side and used instead.
+                    Mutually exclusive with `target_type`/`target_id`.
+                    Requires `view_team` on the team.
+                target_type:
+                  type: string
+                  enum:
+                    - system
+                    - team
+                    - channel
+                  description: >
+                    Single-target scope. Mutually exclusive with `channel_id`
+                    and `team_id`. Required if no hierarchical scope is
+                    given, except when `object_types` is exactly `["system"]`
+                    — in that case any scope or target params are ignored
+                    and the endpoint resolves to `target_type=system`.
+                target_id:
+                  type: string
+                  description: >
+                    Single-target scope. Required when `target_type` is
+                    `channel` or `team`. Mutually exclusive with `channel_id`
+                    and `team_id`.
+                since:
+                  type: integer
+                  format: int64
+                  description: >
+                    Unix timestamp in milliseconds. When greater than 0,
+                    returns fields with `update_at` greater than or equal
+                    to this value, including tombstones.
+                cursor_id:
+                  type: string
+                  description: The ID of the last property field from the previous page, for cursor-based pagination.
+                cursor_create_at:
+                  type: integer
+                  format: int64
+                  description: >
+                    The `create_at` timestamp of the last property field
+                    from the previous page. Required alongside `cursor_id`
+                    when `since` is absent. Mutually exclusive with
+                    `cursor_update_at`.
+                cursor_update_at:
+                  type: integer
+                  format: int64
+                  description: >
+                    The `update_at` timestamp of the last property field
+                    from the previous page. Required alongside `cursor_id`
+                    when `since` is present. Mutually exclusive with
+                    `cursor_create_at`.
+                per_page:
+                  type: integer
+                  default: 60
+                  description: The number of property fields per page.
+      responses:
+        "200":
+          description: Property fields retrieval successful
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/PropertyField"
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "401":
+          $ref: "#/components/responses/Unauthorized"
+        "403":
+          $ref: "#/components/responses/Forbidden"
+        "404":
+          $ref: "#/components/responses/NotFound"
   "/api/v4/properties/groups/{group_name}/{object_type}/fields/{field_id}":
     patch:
       tags:
@@ -281,9 +486,15 @@
         - properties
       summary: Get property values for a target
       description: >
-        Get all property values for a specific target within a group.
+        Get all property values for a specific target within a group. Uses cursor-based pagination.
         The `template` object type cannot have values and will return 400.
         The `system` object type must use the dedicated `/api/v4/properties/groups/{group_name}/system/values` endpoint and will return 400 on this route.
+
+
+        **Delta sync via `since`:** When `since > 0`, the endpoint returns only rows whose `update_at` is greater than the cutoff, *including* tombstoned rows.
+
+
+        **Cursor key must match the active ordering:** in delta mode (`since > 0`) the endpoint orders by `update_at` and pagination requires `cursor_update_at`; otherwise it orders by `create_at` and pagination requires `cursor_create_at`.
       operationId: GetPropertyValues
       parameters:
         - name: group_name
@@ -304,6 +515,15 @@
           required: true
           schema:
             type: string
+        - name: since
+          in: query
+          description: >
+            Unix timestamp in milliseconds. When greater than 0, returns
+            values with `update_at` greater than or equal to this value,
+            including tombstones.
+          schema:
+            type: integer
+            format: int64
         - name: cursor_id
           in: query
           description: The ID of the last property value from the previous page, for cursor-based pagination.
@@ -311,7 +531,19 @@
             type: string
         - name: cursor_create_at
           in: query
-          description: The create_at timestamp of the last property value from the previous page. Must be provided together with cursor_id.
+          description: >
+            The `create_at` timestamp of the last property value from the
+            previous page. Required alongside `cursor_id` when `since` is
+            absent. Mutually exclusive with `cursor_update_at`.
+          schema:
+            type: integer
+            format: int64
+        - name: cursor_update_at
+          in: query
+          description: >
+            The `update_at` timestamp of the last property value from the
+            previous page. Required alongside `cursor_id` when `since` is
+            present. Mutually exclusive with `cursor_create_at`.
           schema:
             type: integer
             format: int64
@@ -407,6 +639,12 @@
         within a group. System-scoped values are readable by any authenticated
         user. This endpoint is the dedicated route for `system` object type;
         the `{object_type}/values/{target_id}` route returns 400 for `system`.
+
+
+        **Delta sync via `since`:** When `since > 0`, the endpoint returns only rows whose `update_at` is greater than the cutoff, *including* tombstones.
+
+
+        **Cursor key must match the active ordering:** in delta mode (`since > 0`) the endpoint orders by `update_at` and pagination requires `cursor_update_at`; otherwise it orders by `create_at` and pagination requires `cursor_create_at`.
       operationId: GetSystemPropertyValues
       parameters:
         - name: group_name
@@ -415,6 +653,15 @@
           required: true
           schema:
             type: string
+        - name: since
+          in: query
+          description: >
+            Unix timestamp in milliseconds. When greater than 0, returns
+            values with `update_at` greater than or equal to this value,
+            including tombstones.
+          schema:
+            type: integer
+            format: int64
         - name: cursor_id
           in: query
           description: The ID of the last property value from the previous page, for cursor-based pagination.
@@ -422,7 +669,19 @@
             type: string
         - name: cursor_create_at
           in: query
-          description: The create_at timestamp of the last property value from the previous page. Must be provided together with cursor_id.
+          description: >
+            The `create_at` timestamp of the last property value from the
+            previous page. Required alongside `cursor_id` when `since` is
+            absent. Mutually exclusive with `cursor_update_at`.
+          schema:
+            type: integer
+            format: int64
+        - name: cursor_update_at
+          in: query
+          description: >
+            The `update_at` timestamp of the last property value from the
+            previous page. Required alongside `cursor_id` when `since` is
+            present. Mutually exclusive with `cursor_create_at`.
           schema:
             type: integer
             format: int64

server/channels/api4/api.go

@@ -176,6 +176,7 @@ type Routes struct {
 	Properties           *mux.Router // 'api/v4/properties'
 	PropertyFields       *mux.Router // 'api/v4/properties/groups/{group_name:[a-z][a-z0-9_]*}/{object_type:[a-z]+}/fields'
 	PropertyField        *mux.Router // 'api/v4/properties/groups/{group_name:[a-z][a-z0-9_]*}/{object_type:[a-z]+}/fields/{field_id:[A-Za-z0-9]+}'
+	PropertyFieldsSearch *mux.Router // 'api/v4/properties/groups/{group_name:[a-z][a-z0-9_]*}/fields/search'
 	PropertyValues       *mux.Router // 'api/v4/properties/groups/{group_name:[a-z][a-z0-9_]*}/{object_type:[a-z]+}/values/{target_id:[A-Za-z0-9]+}'
 	PropertySystemValues *mux.Router // 'api/v4/properties/groups/{group_name:[a-z][a-z0-9_]*}/system/values'
 }
@@ -339,6 +340,7 @@ func Init(srv *app.Server) (*API, error) {
 	api.BaseRoutes.Properties = api.BaseRoutes.APIRoot.PathPrefix("/properties").Subrouter()
 	api.BaseRoutes.PropertyFields = api.BaseRoutes.Properties.PathPrefix("/groups/{group_name:[a-z][a-z0-9_]*}/{object_type:[a-z]+}/fields").Subrouter()
 	api.BaseRoutes.PropertyField = api.BaseRoutes.PropertyFields.PathPrefix("/{field_id:[A-Za-z0-9]+}").Subrouter()
+	api.BaseRoutes.PropertyFieldsSearch = api.BaseRoutes.Properties.PathPrefix("/groups/{group_name:[a-z][a-z0-9_]*}/fields/search").Subrouter()
 	api.BaseRoutes.PropertyValues = api.BaseRoutes.Properties.PathPrefix("/groups/{group_name:[a-z][a-z0-9_]*}/{object_type:[a-z]+}/values/{target_id:[A-Za-z0-9]+}").Subrouter()
 	api.BaseRoutes.PropertySystemValues = api.BaseRoutes.Properties.PathPrefix("/groups/{group_name:[a-z][a-z0-9_]*}/system/values").Subrouter()
 

server/channels/api4/properties.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/v8/channels/app"
+	"github.com/mattermost/mattermost/server/v8/channels/web"
 )
 
 const maxPropertyValuePatchItems = 50
@@ -24,6 +25,7 @@ func (api *API) InitProperties() {
 		api.srv.Config().FeatureFlags.ClassificationMarkings ||
 		api.srv.Config().FeatureFlags.SessionAttributes {
 		api.BaseRoutes.PropertyFields.Handle("", api.APISessionRequired(getPropertyFields)).Methods(http.MethodGet)
+		api.BaseRoutes.PropertyFieldsSearch.Handle("", api.APISessionRequired(searchPropertyFields)).Methods(http.MethodPost)
 		api.BaseRoutes.PropertyValues.Handle("", api.APISessionRequired(getPropertyValues)).Methods(http.MethodGet)
 		api.BaseRoutes.PropertySystemValues.Handle("", api.APISessionRequired(getSystemPropertyValues)).Methods(http.MethodGet)
 
@@ -181,73 +183,151 @@ func getPropertyFields(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	opts := model.PropertyFieldSearchOpts{
+		GroupID:     group.ID,
+		ObjectTypes: []string{c.Params.ObjectType},
+		PerPage:     c.Params.PerPage,
+	}
+
 	query := r.URL.Query()
 
-	// Build search options
-	opts := model.PropertyFieldSearchOpts{
-		GroupID:    group.ID,
-		ObjectType: c.Params.ObjectType,
-		PerPage:    c.Params.PerPage,
+	if s := query.Get("since"); s != "" {
+		since, err := strconv.ParseInt(s, 10, 64)
+		if err != nil {
+			c.SetInvalidParamWithErr("since", err)
+			return
+		}
+		opts.SinceUpdateAt = since
 	}
 
-	// Parse cursor parameters for pagination
+	// Cursor: directory mode uses CreateAt, delta mode (since>0) uses
+	// UpdateAt.
 	if cursorID := query.Get("cursor_id"); cursorID != "" {
-		createAt, _ := strconv.ParseInt(query.Get("cursor_create_at"), 10, 64)
+		cur := model.PropertyFieldSearchCursor{PropertyFieldID: cursorID}
+		if v := query.Get("cursor_update_at"); v != "" {
+			ua, err := strconv.ParseInt(v, 10, 64)
+			if err != nil {
+				c.SetInvalidParamWithErr("cursor_update_at", err)
+				return
+			}
+			cur.UpdateAt = ua
+		}
+		if v := query.Get("cursor_create_at"); v != "" {
+			ca, err := strconv.ParseInt(v, 10, 64)
+			if err != nil {
+				c.SetInvalidParamWithErr("cursor_create_at", err)
+				return
+			}
+			cur.CreateAt = ca
+		}
+		if err := cur.IsValid(); err != nil {
+			c.SetInvalidParamWithErr("cursor", err)
+			return
+		}
+		opts.Cursor = cur
+	}
+
+	opts.ChannelID = query.Get("channel_id")
+	opts.TeamID = query.Get("team_id")
+	opts.TargetType = query.Get("target_type")
+	if t := query.Get("target_id"); t != "" {
+		opts.TargetIDs = []string{t}
+	}
+
+	searchPropertyFieldsCore(c, w, group, opts, "getPropertyFields")
+}
+
+func searchPropertyFields(c *Context, w http.ResponseWriter, r *http.Request) {
+	c.RequireGroupName()
+	if c.Err != nil {
+		return
+	}
+
+	group := getV2Group(c, "searchPropertyFields")
+	if c.Err != nil {
+		return
+	}
+
+	var search model.PropertyFieldSearch
+	if err := json.NewDecoder(r.Body).Decode(&search); err != nil {
+		c.SetInvalidParamWithErr("property_field_search", err)
+		return
+	}
+
+	if len(search.ObjectTypes) == 0 {
+		c.SetInvalidParam("object_types")
+		return
+	}
+	for _, ot := range search.ObjectTypes {
+		if !model.IsValidPropertyFieldObjectType(ot) {
+			c.SetInvalidParam("object_types")
+			return
+		}
+	}
+
+	opts := model.PropertyFieldSearchOpts{
+		GroupID:       group.ID,
+		ObjectTypes:   search.ObjectTypes,
+		ChannelID:     search.ChannelID,
+		TeamID:        search.TeamID,
+		TargetType:    search.TargetType,
+		SinceUpdateAt: search.SinceUpdateAt,
+		PerPage:       search.PerPage,
+	}
+	if search.TargetID != "" {
+		opts.TargetIDs = []string{search.TargetID}
+	}
+	if search.CursorID != "" {
 		opts.Cursor = model.PropertyFieldSearchCursor{
-			PropertyFieldID: cursorID,
-			CreateAt:        createAt,
-		}
-		if err := opts.Cursor.IsValid(); err != nil {
-			c.SetInvalidURLParam("cursor")
-			return
+			PropertyFieldID: search.CursorID,
+			CreateAt:        search.CursorCreateAt,
+			UpdateAt:        search.CursorUpdateAt,
 		}
 	}
-
-	// target_type filter: required in general, but implicit for the system
-	// object type since it can only ever live at the system level.
-	if c.Params.ObjectType == model.PropertyFieldObjectTypeSystem {
-		opts.TargetType = string(model.PropertyFieldTargetLevelSystem)
-	} else {
-		opts.TargetType = query.Get("target_type")
-		if !model.IsValidPSAv2PropertyFieldTargetType(opts.TargetType) {
-			c.Err = model.NewAppError("getPropertyFields", "api.property_field.get.invalid_target_type.app_error", nil, "", http.StatusBadRequest)
-			return
-		}
+	if opts.PerPage <= 0 {
+		opts.PerPage = web.PerPageDefault
+	} else if opts.PerPage > web.PerPageMaximum {
+		opts.PerPage = web.PerPageMaximum
 	}
 
-	// Optional target_id filter
-	if targetID := query.Get("target_id"); targetID != "" {
-		opts.TargetIDs = []string{targetID}
-	}
+	searchPropertyFieldsCore(c, w, group, opts, "searchPropertyFields")
+}
 
+// searchPropertyFieldsCore is the shared pipeline both list-style endpoints
+// share once opts has been populated from query string or request body:
+// scope resolution + permission checks, opts validation, audit, search,
+// response encoding.
+func searchPropertyFieldsCore(c *Context, w http.ResponseWriter, group *model.PropertyGroup, opts model.PropertyFieldSearchOpts, callerName string) {
 	auditRec := c.MakeAuditRecord(model.AuditEventGetPropertyFields, model.AuditStatusFail)
 	defer c.LogAuditRec(auditRec)
 	model.AddEventParameterToAuditRec(auditRec, "group_name", c.Params.GroupName)
-	model.AddEventParameterToAuditRec(auditRec, "object_type", c.Params.ObjectType)
+	model.AddEventParameterToAuditRec(auditRec, "object_types", opts.ObjectTypes)
+	model.AddEventParameterToAuditRec(auditRec, "since", opts.SinceUpdateAt)
+	model.AddEventParameterToAuditRec(auditRec, "channel_id", opts.ChannelID)
+	model.AddEventParameterToAuditRec(auditRec, "team_id", opts.TeamID)
 
-	// Resource-scoped target types require a target_id for access checks
-	switch opts.TargetType {
-	case "channel":
-		if len(opts.TargetIDs) == 0 {
-			c.Err = model.NewAppError("getPropertyFields", "api.property_field.get.target_id_required.app_error", nil, "", http.StatusBadRequest)
-			return
-		}
-		hasPermission, _ := c.App.SessionHasPermissionToChannel(c.AppContext, *c.AppContext.Session(), opts.TargetIDs[0], model.PermissionReadChannel)
-		if !hasPermission {
-			c.SetPermissionError(model.PermissionReadChannel)
-			return
-		}
-	case "team":
-		if len(opts.TargetIDs) == 0 {
-			c.Err = model.NewAppError("getPropertyFields", "api.property_field.get.target_id_required.app_error", nil, "", http.StatusBadRequest)
-			return
-		}
-		if !c.App.SessionHasPermissionToTeam(*c.AppContext.Session(), opts.TargetIDs[0], model.PermissionViewTeam) {
-			c.SetPermissionError(model.PermissionViewTeam)
-			return
-		}
-	case "system":
-		// System-level fields are visible to all authenticated users
+	// System-object fields can only live at the system scope by
+	// invariant (enforced at create time). When the caller asks
+	// exclusively for system-object fields, any channel/team/target
+	// filter is a semantic no-op — we collapse to target_type=system
+	// regardless of what was passed so legacy callers don't get a
+	// confusing scope_conflict on otherwise valid requests. The
+	// shortcut only applies when object_types is exactly [system]:
+	// mixing with other types would silently drop the non-system rows.
+	if len(opts.ObjectTypes) == 1 && opts.ObjectTypes[0] == model.PropertyFieldObjectTypeSystem {
+		opts.ChannelID = ""
+		opts.TeamID = ""
+		opts.TargetIDs = nil
+		opts.TargetType = string(model.PropertyFieldTargetLevelSystem)
+	}
+
+	if !resolveScopeAndCheckPermissions(c, &opts, callerName) {
+		return
+	}
+
+	if err := opts.IsValid(); err != nil {
+		c.Err = model.NewAppError(callerName, "api.property_field.get.invalid_opts.app_error", nil, err.Error(), http.StatusBadRequest)
+		return
 	}
 
 	fields, err := c.App.SearchPropertyFields(c.AppContext, group.ID, opts)
@@ -263,6 +343,75 @@ func getPropertyFields(c *Context, w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// resolveScopeAndCheckPermissions enforces the two scope modes (hierarchical
+// vs single-target) and runs the per-scope permission checks. It operates on
+// already-populated opts.
+func resolveScopeAndCheckPermissions(c *Context, opts *model.PropertyFieldSearchOpts, callerName string) bool {
+	scopeByChanTeam := opts.ChannelID != "" || opts.TeamID != ""
+	scopeByTarget := opts.TargetType != "" || len(opts.TargetIDs) > 0
+	if scopeByChanTeam && scopeByTarget {
+		c.Err = model.NewAppError(callerName, "api.property_field.get.scope_conflict.app_error", nil, "", http.StatusBadRequest)
+		return false
+	}
+
+	switch {
+	case opts.ChannelID != "":
+		hasPermission, _ := c.App.SessionHasPermissionToChannel(c.AppContext, *c.AppContext.Session(), opts.ChannelID, model.PermissionReadChannel)
+		if !hasPermission {
+			c.SetPermissionError(model.PermissionReadChannel)
+			return false
+		}
+		channel, appErr := c.App.GetChannel(c.AppContext, opts.ChannelID)
+		if appErr != nil {
+			c.Err = appErr
+			return false
+		}
+		opts.ChannelID = channel.Id
+		opts.TeamID = channel.TeamId
+	case opts.TeamID != "":
+		if !c.App.SessionHasPermissionToTeam(*c.AppContext.Session(), opts.TeamID, model.PermissionViewTeam) {
+			c.SetPermissionError(model.PermissionViewTeam)
+			return false
+		}
+	case opts.TargetType != "":
+		if !model.IsValidPSAv2PropertyFieldTargetType(opts.TargetType) {
+			c.Err = model.NewAppError(callerName, "api.property_field.get.invalid_target_type.app_error", nil, "", http.StatusBadRequest)
+			return false
+		}
+		switch model.PropertyFieldTargetLevel(opts.TargetType) {
+		case model.PropertyFieldTargetLevelChannel:
+			if len(opts.TargetIDs) == 0 {
+				c.Err = model.NewAppError(callerName, "api.property_field.get.target_id_required.app_error", nil, "", http.StatusBadRequest)
+				return false
+			}
+			hasPermission, _ := c.App.SessionHasPermissionToChannel(c.AppContext, *c.AppContext.Session(), opts.TargetIDs[0], model.PermissionReadChannel)
+			if !hasPermission {
+				c.SetPermissionError(model.PermissionReadChannel)
+				return false
+			}
+		case model.PropertyFieldTargetLevelTeam:
+			if len(opts.TargetIDs) == 0 {
+				c.Err = model.NewAppError(callerName, "api.property_field.get.target_id_required.app_error", nil, "", http.StatusBadRequest)
+				return false
+			}
+			if !c.App.SessionHasPermissionToTeam(*c.AppContext.Session(), opts.TargetIDs[0], model.PermissionViewTeam) {
+				c.SetPermissionError(model.PermissionViewTeam)
+				return false
+			}
+		case model.PropertyFieldTargetLevelSystem:
+			// System-level fields are visible to all authenticated users.
+		}
+	case len(opts.TargetIDs) > 0:
+		// target_id without target_type is malformed.
+		c.Err = model.NewAppError(callerName, "api.property_field.get.target_type_required.app_error", nil, "", http.StatusBadRequest)
+		return false
+	default:
+		c.Err = model.NewAppError(callerName, "api.property_field.get.scope_required.app_error", nil, "", http.StatusBadRequest)
+		return false
+	}
+	return true
+}
+
 func patchPropertyField(c *Context, w http.ResponseWriter, r *http.Request) {
 	c.RequireGroupName().RequireObjectType().RequireFieldId()
 	if c.Err != nil {
@@ -460,12 +609,6 @@ func getPropertyValuesCore(c *Context, w http.ResponseWriter, r *http.Request, o
 		return
 	}
 
-	auditRec := c.MakeAuditRecord(model.AuditEventGetPropertyValues, model.AuditStatusFail)
-	defer c.LogAuditRec(auditRec)
-	model.AddEventParameterToAuditRec(auditRec, "group_name", c.Params.GroupName)
-	model.AddEventParameterToAuditRec(auditRec, "object_type", objectType)
-	model.AddEventParameterToAuditRec(auditRec, "target_id", targetID)
-
 	query := r.URL.Query()
 
 	opts := model.PropertyValueSearchOpts{
@@ -474,19 +617,50 @@ func getPropertyValuesCore(c *Context, w http.ResponseWriter, r *http.Request, o
 		PerPage:    c.Params.PerPage,
 	}
 
-	// Parse cursor parameters for pagination
-	if cursorID := query.Get("cursor_id"); cursorID != "" {
-		createAt, _ := strconv.ParseInt(query.Get("cursor_create_at"), 10, 64)
-		opts.Cursor = model.PropertyValueSearchCursor{
-			PropertyValueID: cursorID,
-			CreateAt:        createAt,
-		}
-		if err := opts.Cursor.IsValid(); err != nil {
-			c.SetInvalidURLParam("cursor")
+	if s := query.Get("since"); s != "" {
+		since, err := strconv.ParseInt(s, 10, 64)
+		if err != nil {
+			c.SetInvalidParamWithErr("since", err)
 			return
 		}
+		opts.SinceUpdateAt = since
 	}
 
+	// Cursor: directory mode uses CreateAt, delta mode (since>0) uses
+	// UpdateAt. opts.IsValid below rejects the mismatched combinations.
+	if cursorID := query.Get("cursor_id"); cursorID != "" {
+		cur := model.PropertyValueSearchCursor{PropertyValueID: cursorID}
+		if v := query.Get("cursor_update_at"); v != "" {
+			ua, err := strconv.ParseInt(v, 10, 64)
+			if err != nil {
+				c.SetInvalidParamWithErr("cursor_update_at", err)
+				return
+			}
+			cur.UpdateAt = ua
+		}
+		if v := query.Get("cursor_create_at"); v != "" {
+			ca, err := strconv.ParseInt(v, 10, 64)
+			if err != nil {
+				c.SetInvalidParamWithErr("cursor_create_at", err)
+				return
+			}
+			cur.CreateAt = ca
+		}
+		opts.Cursor = cur
+	}
+
+	if err := opts.IsValid(); err != nil {
+		c.Err = model.NewAppError("getPropertyValues", "api.property_value.get.invalid_opts.app_error", nil, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	auditRec := c.MakeAuditRecord(model.AuditEventGetPropertyValues, model.AuditStatusFail)
+	defer c.LogAuditRec(auditRec)
+	model.AddEventParameterToAuditRec(auditRec, "group_name", c.Params.GroupName)
+	model.AddEventParameterToAuditRec(auditRec, "object_type", objectType)
+	model.AddEventParameterToAuditRec(auditRec, "target_id", targetID)
+	model.AddEventParameterToAuditRec(auditRec, "since", opts.SinceUpdateAt)
+
 	values, err := c.App.SearchPropertyValues(c.AppContext, group.ID, opts)
 	if err != nil {
 		c.Err = err

server/channels/db/migrations/migrations.list

@@ -395,3 +395,7 @@ channels/db/migrations/postgres/000199_rename_classification_linked_fields.down.
 channels/db/migrations/postgres/000199_rename_classification_linked_fields.up.sql
 channels/db/migrations/postgres/000200_add_rank_to_attribute_view.down.sql
 channels/db/migrations/postgres/000200_add_rank_to_attribute_view.up.sql
+channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.down.sql
+channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.up.sql
+channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.down.sql
+channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.up.sql

server/channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.down.sql

@@ -0,0 +1,2 @@
+-- morph:nontransactional
+DROP INDEX CONCURRENTLY IF EXISTS idx_propertyfields_groupid_updateat_id;

server/channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.up.sql

@@ -0,0 +1,2 @@
+-- morph:nontransactional
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_propertyfields_groupid_updateat_id ON PropertyFields(GroupID, UpdateAt, ID);

server/channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.down.sql

@@ -0,0 +1,2 @@
+-- morph:nontransactional
+DROP INDEX CONCURRENTLY IF EXISTS idx_propertyvalues_groupid_updateat_id;

server/channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.up.sql

@@ -0,0 +1,2 @@
+-- morph:nontransactional
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_propertyvalues_groupid_updateat_id ON PropertyValues(GroupID, UpdateAt, ID);

server/channels/store/sqlstore/property_field_store.go

@@ -180,30 +180,61 @@ func (s *SqlPropertyFieldStore) GetForGroup(ctx context.Context, groupID string)
 	return fields, nil
 }
 
+// SearchPropertyFields runs the PSAv2 field listing query.
+//
+// The store operates in two modes determined by opts.SinceUpdateAt:
+//
+//   - Delta mode (SinceUpdateAt > 0): orders by UpdateAt ASC, Id ASC; paginates
+//     with the (UpdateAt, Id) cursor key; auto-includes soft-deleted rows. The
+//     DeleteAt filter is NOT applied in this mode.
+//   - Directory mode (SinceUpdateAt <= 0): orders by CreateAt ASC, Id ASC;
+//     paginates with the (CreateAt, Id) cursor key; honors opts.IncludeDeleted.
+//
+// The scope filter has two mutually exclusive shapes, enforced by
+// opts.IsValid(): either we search through the hierarchy (channel or team and up,
+// if ChannelID or TeamID are set) or we filter on a single target using
+// TargetType/TargetIDs.
 func (s *SqlPropertyFieldStore) SearchPropertyFields(opts model.PropertyFieldSearchOpts) ([]*model.PropertyField, error) {
-	if err := opts.Cursor.IsValid(); err != nil {
-		return nil, fmt.Errorf("cursor is invalid: %w", err)
+	if err := opts.IsValid(); err != nil {
+		return nil, fmt.Errorf("opts is invalid: %w", err)
 	}
 
 	if opts.PerPage < 1 {
 		return nil, errors.New("per page must be positive integer greater than zero")
 	}
 
-	builder := s.tableSelectQuery.
-		OrderBy("CreateAt ASC, Id ASC").
-		Limit(uint64(opts.PerPage))
+	deltaMode := opts.SinceUpdateAt > 0
 
-	if !opts.Cursor.IsEmpty() {
-		builder = builder.Where(sq.Or{
-			sq.Gt{"CreateAt": opts.Cursor.CreateAt},
-			sq.And{
-				sq.Eq{"CreateAt": opts.Cursor.CreateAt},
-				sq.Gt{"Id": opts.Cursor.PropertyFieldID},
-			},
-		})
+	builder := s.tableSelectQuery.Limit(uint64(opts.PerPage))
+	if deltaMode {
+		builder = builder.OrderBy("UpdateAt ASC, Id ASC")
+	} else {
+		builder = builder.OrderBy("CreateAt ASC, Id ASC")
 	}
 
-	if !opts.IncludeDeleted {
+	if !opts.Cursor.IsEmpty() {
+		if deltaMode {
+			builder = builder.Where(sq.Or{
+				sq.Gt{"UpdateAt": opts.Cursor.UpdateAt},
+				sq.And{
+					sq.Eq{"UpdateAt": opts.Cursor.UpdateAt},
+					sq.Gt{"Id": opts.Cursor.PropertyFieldID},
+				},
+			})
+		} else {
+			builder = builder.Where(sq.Or{
+				sq.Gt{"CreateAt": opts.Cursor.CreateAt},
+				sq.And{
+					sq.Eq{"CreateAt": opts.Cursor.CreateAt},
+					sq.Gt{"Id": opts.Cursor.PropertyFieldID},
+				},
+			})
+		}
+	}
+
+	// Delta mode auto-includes tombstones; directory mode keeps the explicit
+	// IncludeDeleted opt-in.
+	if !deltaMode && !opts.IncludeDeleted {
 		builder = builder.Where(sq.Eq{"DeleteAt": 0})
 	}
 
@@ -211,24 +242,70 @@ func (s *SqlPropertyFieldStore) SearchPropertyFields(opts model.PropertyFieldSea
 		builder = builder.Where(sq.Eq{"GroupID": opts.GroupID})
 	}
 
-	if opts.ObjectType != "" {
+	// Prefer ObjectTypes (renders as IN); fall back to the deprecated
+	// single-value ObjectType for backwards compatibility.
+	if len(opts.ObjectTypes) > 0 {
+		builder = builder.Where(sq.Eq{"ObjectType": opts.ObjectTypes})
+	} else if opts.ObjectType != "" {
 		builder = builder.Where(sq.Eq{"ObjectType": opts.ObjectType})
 	}
 
-	if opts.TargetType != "" {
-		builder = builder.Where(sq.Eq{"TargetType": opts.TargetType})
-	}
-
-	if len(opts.TargetIDs) > 0 {
-		builder = builder.Where(sq.Eq{"TargetID": opts.TargetIDs})
+	// Four mutually exclusive scopes (enforced by opts.IsValid()):
+	//   - Channel + team: OR{system, team=TeamID, channel=ChannelID}
+	//   - Channel only (DM/GM, no team): OR{system, channel=ChannelID}
+	//   - Team-only: OR{system, team=TeamID}
+	//   - Single target: WHERE TargetType = ? and/or TargetID IN (?) — either
+	//     filter may be applied independently for backwards compatibility.
+	switch {
+	case opts.ChannelID != "" && opts.TeamID != "":
+		builder = builder.Where(sq.Or{
+			sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelSystem)},
+			sq.And{
+				sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelTeam)},
+				sq.Eq{"TargetID": opts.TeamID},
+			},
+			sq.And{
+				sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelChannel)},
+				sq.Eq{"TargetID": opts.ChannelID},
+			},
+		})
+	case opts.ChannelID != "":
+		// DM/GM channels have no parent team, so the hierarchy is just
+		// system → channel.
+		builder = builder.Where(sq.Or{
+			sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelSystem)},
+			sq.And{
+				sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelChannel)},
+				sq.Eq{"TargetID": opts.ChannelID},
+			},
+		})
+	case opts.TeamID != "":
+		builder = builder.Where(sq.Or{
+			sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelSystem)},
+			sq.And{
+				sq.Eq{"TargetType": string(model.PropertyFieldTargetLevelTeam)},
+				sq.Eq{"TargetID": opts.TeamID},
+			},
+		})
+	default:
+		if opts.TargetType != "" {
+			builder = builder.Where(sq.Eq{"TargetType": opts.TargetType})
+		}
+		if len(opts.TargetIDs) > 0 {
+			builder = builder.Where(sq.Eq{"TargetID": opts.TargetIDs})
+		}
 	}
 
 	if opts.LinkedFieldID != "" {
 		builder = builder.Where(sq.Eq{"LinkedFieldID": opts.LinkedFieldID})
 	}
 
-	if opts.SinceUpdateAt > 0 {
-		builder = builder.Where(sq.Gt{"UpdateAt": opts.SinceUpdateAt})
+	if deltaMode {
+		// Inclusive boundary so rows updated at exactly `since`
+		// are returned on the first page. The cursor clause above
+		// then disambiguates same-millisecond rows by Id across
+		// subsequent pages.
+		builder = builder.Where(sq.GtOrEq{"UpdateAt": opts.SinceUpdateAt})
 	}
 
 	fields := []*model.PropertyField{}

server/channels/store/sqlstore/property_value_store.go

@@ -135,29 +135,44 @@ func (s *SqlPropertyValueStore) GetMany(groupID string, ids []string) ([]*model.
 }
 
 func (s *SqlPropertyValueStore) SearchPropertyValues(opts model.PropertyValueSearchOpts) ([]*model.PropertyValue, error) {
-	if err := opts.Cursor.IsValid(); err != nil {
-		return nil, fmt.Errorf("cursor is invalid: %w", err)
+	if err := opts.IsValid(); err != nil {
+		return nil, fmt.Errorf("opts is invalid: %w", err)
 	}
 
 	if opts.PerPage < 1 {
 		return nil, errors.New("per page must be positive integer greater than zero")
 	}
 
-	builder := s.tableSelectQuery.
-		OrderBy("CreateAt ASC, Id ASC").
-		Limit(uint64(opts.PerPage))
+	deltaMode := opts.SinceUpdateAt > 0
 
-	if !opts.Cursor.IsEmpty() {
-		builder = builder.Where(sq.Or{
-			sq.Gt{"CreateAt": opts.Cursor.CreateAt},
-			sq.And{
-				sq.Eq{"CreateAt": opts.Cursor.CreateAt},
-				sq.Gt{"Id": opts.Cursor.PropertyValueID},
-			},
-		})
+	builder := s.tableSelectQuery.Limit(uint64(opts.PerPage))
+	if deltaMode {
+		builder = builder.OrderBy("UpdateAt ASC, Id ASC")
+	} else {
+		builder = builder.OrderBy("CreateAt ASC, Id ASC")
 	}
 
-	if !opts.IncludeDeleted {
+	if !opts.Cursor.IsEmpty() {
+		if deltaMode {
+			builder = builder.Where(sq.Or{
+				sq.Gt{"UpdateAt": opts.Cursor.UpdateAt},
+				sq.And{
+					sq.Eq{"UpdateAt": opts.Cursor.UpdateAt},
+					sq.Gt{"Id": opts.Cursor.PropertyValueID},
+				},
+			})
+		} else {
+			builder = builder.Where(sq.Or{
+				sq.Gt{"CreateAt": opts.Cursor.CreateAt},
+				sq.And{
+					sq.Eq{"CreateAt": opts.Cursor.CreateAt},
+					sq.Gt{"Id": opts.Cursor.PropertyValueID},
+				},
+			})
+		}
+	}
+
+	if !deltaMode && !opts.IncludeDeleted {
 		builder = builder.Where(sq.Eq{"DeleteAt": 0})
 	}
 
@@ -177,8 +192,11 @@ func (s *SqlPropertyValueStore) SearchPropertyValues(opts model.PropertyValueSea
 		builder = builder.Where(sq.Eq{"FieldID": opts.FieldID})
 	}
 
-	if opts.SinceUpdateAt > 0 {
-		builder = builder.Where(sq.Gt{"UpdateAt": opts.SinceUpdateAt})
+	if deltaMode {
+		// Inclusive boundary so rows updated at exactly `since` are
+		// returned on the first page. The cursor clause above then
+		// disambiguates same-millisecond rows by Id across pages.
+		builder = builder.Where(sq.GtOrEq{"UpdateAt": opts.SinceUpdateAt})
 	}
 
 	if opts.Value != nil {

server/channels/store/storetest/property_field_store.go

@@ -6,6 +6,7 @@ package storetest
 import (
 	"context"
 	"fmt"
+	"slices"
 	"testing"
 	"time"
 
@@ -24,8 +25,7 @@ func TestPropertyFieldStore(t *testing.T, rctx request.CTX, ss store.Store, s Sq
 	t.Run("GetFieldByName", func(t *testing.T) { testGetFieldByName(t, rctx, ss) })
 	t.Run("UpdatePropertyField", func(t *testing.T) { testUpdatePropertyField(t, rctx, ss) })
 	t.Run("DeletePropertyField", func(t *testing.T) { testDeletePropertyField(t, rctx, ss) })
-	t.Run("SearchPropertyFields", func(t *testing.T) { testSearchPropertyFields(t, rctx, ss) })
-	t.Run("SearchPropertyFieldsSince", func(t *testing.T) { testSearchPropertyFieldsSince(t, rctx, ss) })
+	t.Run("SearchPropertyFields", func(t *testing.T) { testSearchPropertyFields(t, rctx, ss, s) })
 	t.Run("CountForGroup", func(t *testing.T) { testCountForGroup(t, rctx, ss) })
 	t.Run("CheckPropertyNameConflict", func(t *testing.T) { testCheckPropertyNameConflict(t, rctx, ss) })
 	t.Run("CountLinkedFields", func(t *testing.T) { testCountLinkedFields(t, rctx, ss) })
@@ -1106,7 +1106,7 @@ func testCountForGroup(t *testing.T, _ request.CTX, ss store.Store) {
 	})
 }
 
-func testSearchPropertyFields(t *testing.T, _ request.CTX, ss store.Store) {
+func testSearchPropertyFields(t *testing.T, _ request.CTX, ss store.Store, s SqlStore) {
 	groupID := model.NewId()
 	targetID := model.NewId()
 
@@ -1279,41 +1279,48 @@ func testSearchPropertyFields(t *testing.T, _ request.CTX, ss store.Store) {
 			expectedIDs: []string{field1.ID, field2.ID},
 		},
 		{
-			name: "filter by SinceUpdateAt timestamp - no results before",
+			// SinceUpdateAt is inclusive (>=). With since=field5.UpdateAt
+			// the only row in groupID with UpdateAt >= field5.UpdateAt is
+			// field5 itself (delete does not touch UpdateAt, so field4
+			// stays behind in time order).
+			name: "filter by SinceUpdateAt timestamp - returns the boundary row",
 			opts: model.PropertyFieldSearchOpts{
 				GroupID:       groupID,
-				SinceUpdateAt: field5.UpdateAt, // After last active field in groupID
+				SinceUpdateAt: field5.UpdateAt,
 				PerPage:       10,
 			},
-			expectedIDs: []string{},
+			expectedIDs: []string{field5.ID},
 		},
 		{
-			name: "filter by SinceUpdateAt timestamp - get fields after specific time",
+			// Using field1.UpdateAt+1 demonstrates the `>=` boundary excludes
+			// anything strictly before it: field1 is dropped while later rows
+			// (including soft-deleted field4 as a tombstone) are surfaced.
+			name: "filter by SinceUpdateAt timestamp - get fields strictly after field1",
 			opts: model.PropertyFieldSearchOpts{
 				GroupID:       groupID,
-				SinceUpdateAt: field1.UpdateAt, // After field1, should get field2 and field5 from same group
+				SinceUpdateAt: field1.UpdateAt + 1,
 				PerPage:       10,
 			},
-			expectedIDs: []string{field2.ID, field5.ID},
+			expectedIDs: []string{field2.ID, field4.ID, field5.ID},
 		},
 		{
 			name: "filter by SinceUpdateAt timestamp with group filter",
 			opts: model.PropertyFieldSearchOpts{
 				GroupID:       groupID2,
-				SinceUpdateAt: field3.UpdateAt, // After field3, should get field6 from groupID2
+				SinceUpdateAt: field3.UpdateAt, // >= field3, should get field3 + field6 in groupID2
 				PerPage:       10,
 			},
-			expectedIDs: []string{field6.ID},
+			expectedIDs: []string{field3.ID, field6.ID},
 		},
 		{
 			name: "filter by SinceUpdateAt timestamp including deleted",
 			opts: model.PropertyFieldSearchOpts{
 				GroupID:        groupID,
-				SinceUpdateAt:  field2.UpdateAt, // After field2, should get field4 (deleted) and field5
+				SinceUpdateAt:  field2.UpdateAt, // >= field2 in groupID: field2, field4 (deleted), field5
 				IncludeDeleted: true,
 				PerPage:        10,
 			},
-			expectedIDs: []string{field4.ID, field5.ID},
+			expectedIDs: []string{field2.ID, field4.ID, field5.ID},
 		},
 		{
 			name: "filter by ObjectType post",
@@ -1380,110 +1387,514 @@ func testSearchPropertyFields(t *testing.T, _ request.CTX, ss store.Store) {
 			require.ElementsMatch(t, tc.expectedIDs, ids)
 		})
 	}
-}
 
-func testSearchPropertyFieldsSince(t *testing.T, _ request.CTX, ss store.Store) {
-	// Create fields with controlled timestamps for precise testing
-	groupID := model.NewId()
+	t.Run("Since", func(t *testing.T) {
+		// Create fields with controlled timestamps for precise testing
+		groupID := model.NewId()
 
-	// Create field 1 (will remain unchanged)
-	field1, err := ss.PropertyField().Create(&model.PropertyField{
-		GroupID:    groupID,
-		Name:       "Field 1",
-		Type:       model.PropertyFieldTypeText,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-	})
-	require.NoError(t, err)
-
-	time.Sleep(10 * time.Millisecond) // Ensure different timestamps
-
-	// Create field 2 (will be updated later)
-	field2, err := ss.PropertyField().Create(&model.PropertyField{
-		GroupID:    groupID,
-		Name:       "Field 2",
-		Type:       model.PropertyFieldTypeText,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-	})
-	require.NoError(t, err)
-
-	time.Sleep(10 * time.Millisecond)
-
-	// Create field 3 (will remain unchanged)
-	field3, err := ss.PropertyField().Create(&model.PropertyField{
-		GroupID:    groupID,
-		Name:       "Field 3",
-		Type:       model.PropertyFieldTypeText,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-	})
-	require.NoError(t, err)
-
-	// Update field2 to change its UpdateAt timestamp
-	time.Sleep(10 * time.Millisecond)
-	field2.Name = "Field 2 Updated"
-	updatedFields, err := ss.PropertyField().Update("", []*model.PropertyField{field2}, nil)
-	require.NoError(t, err)
-	require.Len(t, updatedFields, 1)
-	updatedField2 := updatedFields[0]
-
-	t.Run("SinceUpdateAt filters correctly by UpdateAt", func(t *testing.T) {
-		// Get fields updated after field1 (should get field2 and field3)
-		results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: field1.UpdateAt,
-			PerPage:       10,
+		// Create field 1 (will remain unchanged)
+		field1, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID:    groupID,
+			Name:       "Field 1",
+			Type:       model.PropertyFieldTypeText,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
 		})
 		require.NoError(t, err)
-		require.Len(t, results, 2)
 
-		resultIDs := make([]string, len(results))
-		for i, result := range results {
-			resultIDs[i] = result.ID
+		time.Sleep(10 * time.Millisecond) // Ensure different timestamps
+
+		// Create field 2 (will be updated later)
+		field2, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID:    groupID,
+			Name:       "Field 2",
+			Type:       model.PropertyFieldTypeText,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+		})
+		require.NoError(t, err)
+
+		time.Sleep(10 * time.Millisecond)
+
+		// Create field 3 (will remain unchanged)
+		field3, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID:    groupID,
+			Name:       "Field 3",
+			Type:       model.PropertyFieldTypeText,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+		})
+		require.NoError(t, err)
+
+		// Update field2 to change its UpdateAt timestamp
+		time.Sleep(10 * time.Millisecond)
+		field2.Name = "Field 2 Updated"
+		updatedFields, err := ss.PropertyField().Update("", []*model.PropertyField{field2}, nil)
+		require.NoError(t, err)
+		require.Len(t, updatedFields, 1)
+		updatedField2 := updatedFields[0]
+
+		t.Run("SinceUpdateAt filters correctly by UpdateAt", func(t *testing.T) {
+			// Get fields updated at-or-after field1: with `>=` semantics
+			// field1 itself is included, plus field3 and the
+			// post-update field2.
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: field1.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 3)
+
+			resultIDs := make([]string, len(results))
+			for i, result := range results {
+				resultIDs[i] = result.ID
+			}
+			require.ElementsMatch(t, []string{field1.ID, field2.ID, field3.ID}, resultIDs)
+		})
+
+		t.Run("SinceUpdateAt with boundary condition", func(t *testing.T) {
+			// Get fields updated after just before field3's timestamp
+			// Should get both field3 and field2 (which was updated last and now has the most recent UpdateAt), so expect 2 results
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: field3.UpdateAt - 1, // Slightly before field3's timestamp
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 2)
+
+			resultIDs := make([]string, len(results))
+			for i, result := range results {
+				resultIDs[i] = result.ID
+			}
+			// Should get both field2 (updated with new timestamp) and field3
+			require.ElementsMatch(t, []string{field2.ID, field3.ID}, resultIDs)
+		})
+
+		t.Run("SinceUpdateAt at the most recent update returns just that row", func(t *testing.T) {
+			// `>=` semantics: querying at the highest UpdateAt in the
+			// group returns the row at exactly that timestamp.
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: updatedField2.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 1)
+			require.Equal(t, updatedField2.ID, results[0].ID)
+		})
+
+		t.Run("SinceUpdateAt with very recent timestamp", func(t *testing.T) {
+			// Get fields updated since current time
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: model.GetMillis(),
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 0)
+		})
+
+		t.Run("same-millisecond rows are paged correctly via (UpdateAt, Id) cursor", func(t *testing.T) {
+			// Three rows must share the same UpdateAt to exercise the
+			// disambiguation clause. The Update store call pins UpdateAt
+			// to a single GetMillis() value across all rows in the batch.
+			tieGroup := model.NewId()
+			tieFields := make([]*model.PropertyField, 0, 3)
+			for range 3 {
+				f, cerr := ss.PropertyField().Create(&model.PropertyField{
+					GroupID:    tieGroup,
+					Name:       model.NewId(),
+					Type:       model.PropertyFieldTypeText,
+					TargetID:   model.NewId(),
+					TargetType: "test_type",
+				})
+				require.NoError(t, cerr)
+				tieFields = append(tieFields, f)
+				time.Sleep(2 * time.Millisecond)
+			}
+
+			// Bulk-update all three to give them the SAME UpdateAt.
+			for _, f := range tieFields {
+				f.Name = f.Name + "-bumped"
+			}
+			updated, err := ss.PropertyField().Update("", tieFields, nil)
+			require.NoError(t, err)
+			require.Len(t, updated, 3)
+			tieUpdateAt := updated[0].UpdateAt
+			require.Equal(t, tieUpdateAt, updated[1].UpdateAt)
+			require.Equal(t, tieUpdateAt, updated[2].UpdateAt)
+
+			// Page 1: include the boundary row at exactly tieUpdateAt.
+			page1, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       tieGroup,
+				SinceUpdateAt: tieUpdateAt,
+				PerPage:       2,
+			})
+			require.NoError(t, err)
+			require.Len(t, page1, 2, "boundary row + one more must come back on the first page")
+
+			// Page 2: cursor with the last row of page 1 must surface
+			// the third tied row — proving (UpdateAt, Id) disambiguates.
+			last := page1[len(page1)-1]
+			page2, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       tieGroup,
+				SinceUpdateAt: tieUpdateAt,
+				Cursor: model.PropertyFieldSearchCursor{
+					PropertyFieldID: last.ID,
+					UpdateAt:        last.UpdateAt,
+				},
+				PerPage: 2,
+			})
+			require.NoError(t, err)
+			require.Len(t, page2, 1)
+
+			// All three tied rows surfaced exactly once across both pages.
+			seen := map[string]bool{}
+			for _, r := range append(page1, page2...) {
+				seen[r.ID] = true
+			}
+			require.Len(t, seen, 3)
+			for _, f := range tieFields {
+				require.True(t, seen[f.ID], "all tied rows must be returned exactly once across pagination")
+			}
+		})
+	})
+
+	t.Run("Scope", func(t *testing.T) {
+		groupID := model.NewId()
+		teamA := model.NewId()
+		teamB := model.NewId()
+		channelX := model.NewId()
+		channelY := model.NewId()
+
+		// Mixed fixtures: 1 system, 1 team-A, 1 team-B, 2 channel-X (team-A scoped),
+		// 1 channel-Y (team-A scoped). ObjectType varies to exercise the IN filter.
+		systemField := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "system-field",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem),
+			ObjectType: model.PropertyFieldObjectTypeSystem,
 		}
-		require.ElementsMatch(t, []string{field2.ID, field3.ID}, resultIDs)
-	})
-
-	t.Run("SinceUpdateAt with boundary condition", func(t *testing.T) {
-		// Get fields updated after just before field3's timestamp
-		// Should get both field3 and field2 (which was updated last and now has the most recent UpdateAt), so expect 2 results
-		results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: field3.UpdateAt - 1, // Slightly before field3's timestamp
-			PerPage:       10,
-		})
-		require.NoError(t, err)
-		require.Len(t, results, 2)
-
-		resultIDs := make([]string, len(results))
-		for i, result := range results {
-			resultIDs[i] = result.ID
+		teamAField := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "team-a-field",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelTeam),
+			TargetID:   teamA,
+			ObjectType: model.PropertyFieldObjectTypeChannel,
 		}
-		// Should get both field2 (updated with new timestamp) and field3
-		require.ElementsMatch(t, []string{field2.ID, field3.ID}, resultIDs)
+		teamBField := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "team-b-field",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelTeam),
+			TargetID:   teamB,
+			ObjectType: model.PropertyFieldObjectTypeChannel,
+		}
+		channelXField1 := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "channel-x-field-1",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelChannel),
+			TargetID:   channelX,
+			ObjectType: model.PropertyFieldObjectTypeChannel,
+		}
+		channelXField2 := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "channel-x-field-2",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelChannel),
+			TargetID:   channelX,
+			ObjectType: model.PropertyFieldObjectTypeUser,
+		}
+		channelYField := &model.PropertyField{
+			GroupID:    groupID,
+			Name:       "channel-y-field",
+			Type:       model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelChannel),
+			TargetID:   channelY,
+			ObjectType: model.PropertyFieldObjectTypeChannel,
+		}
+
+		for _, f := range []*model.PropertyField{systemField, teamAField, teamBField, channelXField1, channelXField2, channelYField} {
+			_, err := ss.PropertyField().Create(f)
+			require.NoError(t, err)
+			time.Sleep(2 * time.Millisecond)
+		}
+
+		t.Run("team-only scope returns system + team-A only", func(t *testing.T) {
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID: groupID,
+				TeamID:  teamA,
+				PerPage: 50,
+			})
+			require.NoError(t, err)
+
+			ids := make([]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+			}
+			require.ElementsMatch(t, []string{systemField.ID, teamAField.ID}, ids)
+		})
+
+		t.Run("channel + team scope returns system + team-A + both channel-X rows", func(t *testing.T) {
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:   groupID,
+				TeamID:    teamA,
+				ChannelID: channelX,
+				PerPage:   50,
+			})
+			require.NoError(t, err)
+
+			ids := make([]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+			}
+			require.ElementsMatch(t,
+				[]string{systemField.ID, teamAField.ID, channelXField1.ID, channelXField2.ID},
+				ids,
+			)
+		})
+
+		t.Run("ObjectTypes IN list returns rows of both kinds", func(t *testing.T) {
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:     groupID,
+				ObjectTypes: []string{model.PropertyFieldObjectTypeChannel, model.PropertyFieldObjectTypeSystem},
+				PerPage:     50,
+			})
+			require.NoError(t, err)
+
+			ids := make([]string, len(results))
+			objectTypes := make(map[string]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+				objectTypes[f.ID] = f.ObjectType
+			}
+			// channelXField2 (user) and any non-channel/non-system rows must be absent.
+			require.ElementsMatch(t,
+				[]string{systemField.ID, teamAField.ID, teamBField.ID, channelXField1.ID, channelYField.ID},
+				ids,
+			)
+			for _, ot := range objectTypes {
+				require.Contains(t,
+					[]string{model.PropertyFieldObjectTypeChannel, model.PropertyFieldObjectTypeSystem},
+					ot,
+					"unexpected object_type in IN-list result",
+				)
+			}
+		})
+
+		t.Run("ObjectType=system combined with ChannelID/TeamID still surfaces system rows", func(t *testing.T) {
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:    groupID,
+				ObjectType: model.PropertyFieldObjectTypeSystem,
+				TeamID:     teamA,
+				ChannelID:  channelX,
+				PerPage:    50,
+			})
+			require.NoError(t, err)
+
+			// teamAField/channelXField* are not ObjectType=system, so they're filtered
+			// out by the ObjectType clause. The OR scope clause still admits system rows
+			// — the regression we're guarding against is the channel/team filter
+			// accidentally excluding the system row.
+			ids := make([]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+			}
+			require.ElementsMatch(t, []string{systemField.ID}, ids)
+		})
+
+		t.Run("ChannelID without TeamID returns system + channel rows (DM/GM scope)", func(t *testing.T) {
+			// DM/GM channels have no parent team, so the hierarchy
+			// collapses to system → channel. Team-scoped rows must not
+			// leak in even though channelX itself happens to be in
+			// teamA in this fixture.
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:   groupID,
+				ChannelID: channelX,
+				PerPage:   50,
+			})
+			require.NoError(t, err)
+
+			ids := make([]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+			}
+			require.ElementsMatch(t,
+				[]string{systemField.ID, channelXField1.ID, channelXField2.ID},
+				ids,
+			)
+		})
+
+		t.Run("scope conflict (TeamID + TargetType) is rejected by IsValid", func(t *testing.T) {
+			_, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:    groupID,
+				TeamID:     teamA,
+				TargetType: string(model.PropertyFieldTargetLevelChannel),
+				PerPage:    50,
+			})
+			require.Error(t, err)
+			require.ErrorContains(t, err, "cannot be combined")
+		})
 	})
 
-	t.Run("SinceUpdateAt after all updates", func(t *testing.T) {
-		// Get fields updated after the most recent update
-		results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: updatedField2.UpdateAt, // After the update
-			PerPage:       10,
+	t.Run("DeltaTombstones", func(t *testing.T) {
+		// Covers the rule that SinceUpdateAt > 0 auto-includes soft-deleted rows,
+		// while SinceUpdateAt == 0 (and the param being absent entirely) still
+		// excludes them — and that the two "unfiltered" calls behave identically.
+		groupID := model.NewId()
+
+		// Pin the boundary at least one ms before the first create so all three
+		// fields are guaranteed to satisfy UpdateAt > beforeAnyCreate. Without
+		// the sleep the first create can land on the same millisecond and the
+		// strict-greater filter would silently exclude it.
+		beforeAnyCreate := model.GetMillis()
+		time.Sleep(2 * time.Millisecond)
+
+		// Two live fields and one we will tombstone.
+		live1, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "live-1", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
 		})
 		require.NoError(t, err)
-		require.Len(t, results, 0) // Should be empty
+		time.Sleep(5 * time.Millisecond)
+
+		live2, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "live-2", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
+		})
+		require.NoError(t, err)
+		time.Sleep(5 * time.Millisecond)
+
+		tombstoned, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "tombstoned", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
+		})
+		require.NoError(t, err)
+		time.Sleep(5 * time.Millisecond)
+
+		require.NoError(t, ss.PropertyField().Delete("", tombstoned.ID))
+
+		t.Run("since > 0 returns UpdateAt > since including soft-deleted rows", func(t *testing.T) {
+			results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: beforeAnyCreate,
+				PerPage:       50,
+			})
+			require.NoError(t, err)
+
+			ids := make([]string, len(results))
+			for i, f := range results {
+				ids[i] = f.ID
+			}
+			require.ElementsMatch(t, []string{live1.ID, live2.ID, tombstoned.ID}, ids)
+		})
+
+		t.Run("since=0 and since absent behave identically and exclude tombstones", func(t *testing.T) {
+			withZeroSince, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: 0,
+				PerPage:       50,
+			})
+			require.NoError(t, err)
+
+			withNoSince, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID: groupID,
+				PerPage: 50,
+			})
+			require.NoError(t, err)
+
+			zeroIDs := make([]string, len(withZeroSince))
+			for i, f := range withZeroSince {
+				zeroIDs[i] = f.ID
+			}
+			noIDs := make([]string, len(withNoSince))
+			for i, f := range withNoSince {
+				noIDs[i] = f.ID
+			}
+
+			// Both calls must exclude the tombstoned row and match each other.
+			require.ElementsMatch(t, []string{live1.ID, live2.ID}, zeroIDs)
+			require.ElementsMatch(t, zeroIDs, noIDs)
+		})
 	})
 
-	t.Run("SinceUpdateAt with very recent timestamp", func(t *testing.T) {
-		// Get fields updated since current time
-		results, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: model.GetMillis(),
-			PerPage:       10,
+	t.Run("DeltaCursor", func(t *testing.T) {
+		// Covers paginated iteration in delta mode when multiple fields share a
+		// single UpdateAt millisecond — the (UpdateAt, Id) cursor tiebreaker must
+		// return every row exactly once, in Id ASC order within the tied bucket.
+		// We pin UpdateAt directly via SQL because Go's wall clock has
+		// sub-millisecond resolution and three model.GetMillis() calls would
+		// generally produce three distinct values.
+		groupID := model.NewId()
+
+		beforeBucket := model.GetMillis()
+		time.Sleep(5 * time.Millisecond)
+
+		f1, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "tied-1", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
 		})
 		require.NoError(t, err)
-		require.Len(t, results, 0)
+		f2, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "tied-2", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
+		})
+		require.NoError(t, err)
+		f3, err := ss.PropertyField().Create(&model.PropertyField{
+			GroupID: groupID, Name: "tied-3", Type: model.PropertyFieldTypeText,
+			TargetType: string(model.PropertyFieldTargetLevelSystem), ObjectType: model.PropertyFieldObjectTypeSystem,
+		})
+		require.NoError(t, err)
+
+		// Force the three rows to share a single UpdateAt value above beforeBucket.
+		// Without this, postgres millisecond resolution still usually produces
+		// distinct timestamps and the tiebreaker would never fire.
+		sharedUpdateAt := model.GetMillis() + 1
+		updateQuery, updateArgs, err := sq.StatementBuilder.PlaceholderFormat(s.GetQueryPlaceholder()).
+			Update("PropertyFields").
+			Set("UpdateAt", sharedUpdateAt).
+			Where(sq.Eq{"Id": []string{f1.ID, f2.ID, f3.ID}}).
+			ToSql()
+		require.NoError(t, err)
+		_, execErr := s.GetMaster().Exec(updateQuery, updateArgs...)
+		require.NoError(t, execErr)
+
+		// Expected order in delta mode: UpdateAt ASC, then Id ASC. All three share
+		// UpdateAt, so the final order is purely Id ASC.
+		expectedIDsSorted := []string{f1.ID, f2.ID, f3.ID}
+		slices.Sort(expectedIDsSorted)
+
+		// Paginate per_page=1 across the boundary; we must walk all three rows
+		// exactly once, never duplicate, never skip.
+		collected := []string{}
+		cursor := model.PropertyFieldSearchCursor{}
+		for range 5 { // hard cap to avoid runaway loop if the cursor never advances
+			batch, err := ss.PropertyField().SearchPropertyFields(model.PropertyFieldSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: beforeBucket,
+				Cursor:        cursor,
+				PerPage:       1,
+			})
+			require.NoError(t, err)
+			if len(batch) == 0 {
+				break
+			}
+			require.Len(t, batch, 1)
+			collected = append(collected, batch[0].ID)
+			cursor = model.PropertyFieldSearchCursor{
+				PropertyFieldID: batch[0].ID,
+				UpdateAt:        batch[0].UpdateAt,
+			}
+		}
+
+		require.Equal(t, expectedIDsSorted, collected,
+			"per_page=1 walk across a tied UpdateAt bucket must return each row exactly once in Id ASC order",
+		)
 	})
 }
 

server/channels/store/storetest/property_value_store.go

@@ -27,7 +27,6 @@ func TestPropertyValueStore(t *testing.T, rctx request.CTX, ss store.Store, s Sq
 	t.Run("UpsertPropertyValue", func(t *testing.T) { testUpsertPropertyValue(t, rctx, ss, s) })
 	t.Run("DeletePropertyValue", func(t *testing.T) { testDeletePropertyValue(t, rctx, ss) })
 	t.Run("SearchPropertyValues", func(t *testing.T) { testSearchPropertyValues(t, rctx, ss, s) })
-	t.Run("SearchPropertyValuesSince", func(t *testing.T) { testSearchPropertyValuesSince(t, rctx, ss) })
 	t.Run("DeleteForField", func(t *testing.T) { testDeleteForField(t, rctx, ss) })
 	t.Run("DeleteForTarget", func(t *testing.T) { testDeleteForTarget(t, rctx, ss) })
 }
@@ -1375,38 +1374,35 @@ func testSearchPropertyValues(t *testing.T, _ request.CTX, ss store.Store, s Sql
 			expectedIDs: []string{value1.ID, value2.ID},
 		},
 		{
-			name: "filter by SinceUpdateAt timestamp - no results before",
+			// Delta mode auto-includes tombstones so the at-or-after-value3
+			// cutoff surfaces value4 even after it has been soft-deleted.
+			// With `>=` semantics value3 itself is also included.
+			name: "filter by SinceUpdateAt timestamp - includes tombstones automatically",
 			opts: model.PropertyValueSearchOpts{
-				SinceUpdateAt: value3.UpdateAt, // After all existing values
+				SinceUpdateAt: value3.UpdateAt,
 				PerPage:       10,
 			},
-			expectedIDs: []string{},
+			expectedIDs: []string{value3.ID, value4.ID},
 		},
 		{
-			name: "filter by SinceUpdateAt timestamp - get values after specific time",
+			// Using value1.UpdateAt+1 demonstrates the `>=` boundary excludes
+			// anything strictly before it: value1 is dropped while later rows
+			// (including soft-deleted value4) are surfaced.
+			name: "filter by SinceUpdateAt timestamp - get values strictly after value1",
 			opts: model.PropertyValueSearchOpts{
-				SinceUpdateAt: value1.UpdateAt, // After value1, should get value2 and value3
+				SinceUpdateAt: value1.UpdateAt + 1,
 				PerPage:       10,
 			},
-			expectedIDs: []string{value2.ID, value3.ID},
+			expectedIDs: []string{value2.ID, value3.ID, value4.ID},
 		},
 		{
 			name: "filter by SinceUpdateAt timestamp with group filter",
 			opts: model.PropertyValueSearchOpts{
 				GroupID:       groupID,
-				SinceUpdateAt: value1.UpdateAt, // After value1, should only get value2 from same group
+				SinceUpdateAt: value1.UpdateAt + 1,
 				PerPage:       10,
 			},
-			expectedIDs: []string{value2.ID},
-		},
-		{
-			name: "filter by SinceUpdateAt timestamp including deleted",
-			opts: model.PropertyValueSearchOpts{
-				SinceUpdateAt:  value3.UpdateAt, // After value3, should get value4 (deleted)
-				IncludeDeleted: true,
-				PerPage:        10,
-			},
-			expectedIDs: []string{value4.ID},
+			expectedIDs: []string{value2.ID, value4.ID},
 		},
 	}
 
@@ -1440,110 +1436,312 @@ func testSearchPropertyValues(t *testing.T, _ request.CTX, ss store.Store, s Sql
 		require.Empty(t, results[0].CreatedBy)
 		require.Empty(t, results[0].UpdatedBy)
 	})
-}
 
-func testSearchPropertyValuesSince(t *testing.T, _ request.CTX, ss store.Store) {
-	// Create values with controlled timestamps for precise testing
-	groupID := model.NewId()
+	t.Run("Since", func(t *testing.T) {
+		// Controlled fixtures: value1 (untouched), value2 (Update bumps its
+		// UpdateAt past value3's), value3 (untouched).
+		groupID := model.NewId()
 
-	// Create value 1 (will remain unchanged)
-	value1, err := ss.PropertyValue().Create(&model.PropertyValue{
-		GroupID:    groupID,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-		FieldID:    model.NewId(),
-		Value:      json.RawMessage(`"value1"`),
-	})
-	require.NoError(t, err)
-
-	time.Sleep(10 * time.Millisecond) // Ensure different timestamps
-
-	// Create value 2 (will be updated later)
-	value2, err := ss.PropertyValue().Create(&model.PropertyValue{
-		GroupID:    groupID,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-		FieldID:    model.NewId(),
-		Value:      json.RawMessage(`"value2"`),
-	})
-	require.NoError(t, err)
-
-	time.Sleep(10 * time.Millisecond)
-
-	// Create value 3 (will remain unchanged)
-	value3, err := ss.PropertyValue().Create(&model.PropertyValue{
-		GroupID:    groupID,
-		TargetID:   model.NewId(),
-		TargetType: "test_type",
-		FieldID:    model.NewId(),
-		Value:      json.RawMessage(`"value3"`),
-	})
-	require.NoError(t, err)
-
-	// Update value2 to change its UpdateAt timestamp
-	time.Sleep(10 * time.Millisecond)
-	value2.Value = json.RawMessage(`"value2_updated"`)
-	updatedValues, err := ss.PropertyValue().Update("", []*model.PropertyValue{value2})
-	require.NoError(t, err)
-	require.Len(t, updatedValues, 1)
-	updatedValue2 := updatedValues[0]
-
-	t.Run("SinceUpdateAt filters correctly by UpdateAt", func(t *testing.T) {
-		// Get values updated after value1 (should get value2 and value3)
-		results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: value1.UpdateAt,
-			PerPage:       10,
+		value1, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"value1"`),
 		})
 		require.NoError(t, err)
-		require.Len(t, results, 2)
+		time.Sleep(10 * time.Millisecond)
 
-		resultIDs := make([]string, len(results))
-		for i, result := range results {
-			resultIDs[i] = result.ID
+		value2, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"value2"`),
+		})
+		require.NoError(t, err)
+		time.Sleep(10 * time.Millisecond)
+
+		value3, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"value3"`),
+		})
+		require.NoError(t, err)
+
+		time.Sleep(10 * time.Millisecond)
+		value2.Value = json.RawMessage(`"value2_updated"`)
+		updatedValues, err := ss.PropertyValue().Update("", []*model.PropertyValue{value2})
+		require.NoError(t, err)
+		require.Len(t, updatedValues, 1)
+		updatedValue2 := updatedValues[0]
+
+		t.Run("SinceUpdateAt filters correctly by UpdateAt", func(t *testing.T) {
+			// `>=` semantics: value1 is included at the boundary,
+			// plus value3 and the post-update value2.
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 3)
+
+			resultIDs := make([]string, len(results))
+			for i, result := range results {
+				resultIDs[i] = result.ID
+			}
+			require.ElementsMatch(t, []string{value1.ID, value2.ID, value3.ID}, resultIDs)
+		})
+
+		t.Run("SinceUpdateAt with boundary condition", func(t *testing.T) {
+			// `value3.UpdateAt - 1` keeps value3 in the window and value2's
+			// post-Update timestamp is even later, so both are returned.
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value3.UpdateAt - 1,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 2)
+
+			resultIDs := make([]string, len(results))
+			for i, result := range results {
+				resultIDs[i] = result.ID
+			}
+			require.ElementsMatch(t, []string{value2.ID, value3.ID}, resultIDs)
+		})
+
+		t.Run("SinceUpdateAt at the most recent update returns just that row", func(t *testing.T) {
+			// `>=` semantics: querying at the highest UpdateAt in the
+			// group returns the row at exactly that timestamp.
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: updatedValue2.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 1)
+			require.Equal(t, updatedValue2.ID, results[0].ID)
+		})
+
+		t.Run("SinceUpdateAt with very recent timestamp", func(t *testing.T) {
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: model.GetMillis(),
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 0)
+		})
+
+		t.Run("same-millisecond rows are paged correctly via (UpdateAt, Id) cursor", func(t *testing.T) {
+			// Three rows share an UpdateAt (Update assigns one
+			// GetMillis() to the whole batch) so the cursor must use
+			// (UpdateAt, Id) to disambiguate.
+			tieGroup := model.NewId()
+			tieValues := make([]*model.PropertyValue, 0, 3)
+			for i := range 3 {
+				v, cerr := ss.PropertyValue().Create(&model.PropertyValue{
+					GroupID:    tieGroup,
+					TargetID:   model.NewId(),
+					TargetType: "test_type",
+					FieldID:    model.NewId(),
+					Value:      json.RawMessage(`"v` + string(rune('0'+i)) + `"`),
+				})
+				require.NoError(t, cerr)
+				tieValues = append(tieValues, v)
+				time.Sleep(2 * time.Millisecond)
+			}
+
+			for _, v := range tieValues {
+				v.Value = json.RawMessage(`"bumped"`)
+			}
+			updated, err := ss.PropertyValue().Update("", tieValues)
+			require.NoError(t, err)
+			require.Len(t, updated, 3)
+			tieUpdateAt := updated[0].UpdateAt
+			require.Equal(t, tieUpdateAt, updated[1].UpdateAt)
+			require.Equal(t, tieUpdateAt, updated[2].UpdateAt)
+
+			page1, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       tieGroup,
+				SinceUpdateAt: tieUpdateAt,
+				PerPage:       2,
+			})
+			require.NoError(t, err)
+			require.Len(t, page1, 2, "boundary row + one more must come back on the first page")
+
+			last := page1[len(page1)-1]
+			page2, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       tieGroup,
+				SinceUpdateAt: tieUpdateAt,
+				Cursor: model.PropertyValueSearchCursor{
+					PropertyValueID: last.ID,
+					UpdateAt:        last.UpdateAt,
+				},
+				PerPage: 2,
+			})
+			require.NoError(t, err)
+			require.Len(t, page2, 1)
+
+			seen := map[string]bool{}
+			for _, r := range append(page1, page2...) {
+				seen[r.ID] = true
+			}
+			require.Len(t, seen, 3)
+			for _, v := range tieValues {
+				require.True(t, seen[v.ID], "all tied rows must be returned exactly once across pagination")
+			}
+		})
+	})
+
+	t.Run("DeltaMode", func(t *testing.T) {
+		// Fresh fixtures so the tombstone subtest does not poison the Since
+		// block above. value2 is later Update'd to push it past value3, then
+		// value3 is soft-deleted to verify tombstone auto-inclusion.
+		groupID := model.NewId()
+
+		value1, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"v1"`),
+		})
+		require.NoError(t, err)
+		time.Sleep(10 * time.Millisecond)
+
+		value2, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"v2"`),
+		})
+		require.NoError(t, err)
+		time.Sleep(10 * time.Millisecond)
+
+		value3, err := ss.PropertyValue().Create(&model.PropertyValue{
+			GroupID:    groupID,
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"v3"`),
+		})
+		require.NoError(t, err)
+
+		time.Sleep(10 * time.Millisecond)
+		value2.Value = json.RawMessage(`"v2-updated"`)
+		updated, err := ss.PropertyValue().Update("", []*model.PropertyValue{value2})
+		require.NoError(t, err)
+		require.Len(t, updated, 1)
+		value2 = updated[0]
+
+		idsOf := func(values []*model.PropertyValue) []string {
+			ids := make([]string, len(values))
+			for i, v := range values {
+				ids[i] = v.ID
+			}
+			return ids
 		}
-		require.ElementsMatch(t, []string{value2.ID, value3.ID}, resultIDs)
-	})
 
-	t.Run("SinceUpdateAt with boundary condition", func(t *testing.T) {
-		// Get values updated after value3's timestamp
-		// Should get both value2 (updated) and value3, so expect 2 results
-		results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: value3.UpdateAt - 1, // Slightly before value3's timestamp
-			PerPage:       10,
+		t.Run("orders by UpdateAt ASC, Id ASC", func(t *testing.T) {
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			// `>=` semantics: value1 is included at the boundary, then
+			// value3 (UpdateAt=t3), then value2 (UpdateAt=t4, bumped
+			// by Update).
+			require.Equal(t, []string{value1.ID, value3.ID, value2.ID}, idsOf(results))
 		})
-		require.NoError(t, err)
-		require.Len(t, results, 2)
 
-		resultIDs := make([]string, len(results))
-		for i, result := range results {
-			resultIDs[i] = result.ID
-		}
-		// Should get both value2 (updated with new timestamp) and value3
-		require.ElementsMatch(t, []string{value2.ID, value3.ID}, resultIDs)
-	})
+		t.Run("auto-includes tombstones", func(t *testing.T) {
+			require.NoError(t, ss.PropertyValue().Delete("", value3.ID))
 
-	t.Run("SinceUpdateAt after all updates", func(t *testing.T) {
-		// Get values updated after the most recent update
-		results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: updatedValue2.UpdateAt, // After the update
-			PerPage:       10,
+			results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				PerPage:       10,
+			})
+			require.NoError(t, err)
+			// value3 must remain visible after soft-delete — the client
+			// needs the tombstone to apply locally. Delete only sets
+			// DeleteAt, so its UpdateAt is unchanged and still
+			// precedes value2. `>=` adds value1 at the boundary.
+			require.Equal(t, []string{value1.ID, value3.ID, value2.ID}, idsOf(results))
+			for _, r := range results {
+				if r.ID == value3.ID {
+					require.NotZero(t, r.DeleteAt, "value3 should be returned with DeleteAt set")
+				}
+			}
 		})
-		require.NoError(t, err)
-		require.Len(t, results, 0) // Should be empty
-	})
 
-	t.Run("SinceUpdateAt with very recent timestamp", func(t *testing.T) {
-		// Get values updated since current time
-		results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
-			GroupID:       groupID,
-			SinceUpdateAt: model.GetMillis(),
-			PerPage:       10,
+		t.Run("paginates with cursor UpdateAt", func(t *testing.T) {
+			// `>=` semantics: first page is value1 (the boundary row),
+			// cursored to value3, then value2.
+			first, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				PerPage:       1,
+			})
+			require.NoError(t, err)
+			require.Equal(t, []string{value1.ID}, idsOf(first))
+
+			second, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				Cursor: model.PropertyValueSearchCursor{
+					PropertyValueID: first[0].ID,
+					UpdateAt:        first[0].UpdateAt,
+				},
+				PerPage: 1,
+			})
+			require.NoError(t, err)
+			require.Equal(t, []string{value3.ID}, idsOf(second))
+
+			third, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				Cursor: model.PropertyValueSearchCursor{
+					PropertyValueID: second[0].ID,
+					UpdateAt:        second[0].UpdateAt,
+				},
+				PerPage: 1,
+			})
+			require.NoError(t, err)
+			require.Equal(t, []string{value2.ID}, idsOf(third))
+		})
+
+		t.Run("directory mode rejects cursor_update_at", func(t *testing.T) {
+			_, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID: groupID,
+				Cursor: model.PropertyValueSearchCursor{
+					PropertyValueID: value1.ID,
+					UpdateAt:        value1.UpdateAt,
+				},
+				PerPage: 10,
+			})
+			require.Error(t, err)
+		})
+
+		t.Run("delta mode rejects cursor_create_at", func(t *testing.T) {
+			_, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+				GroupID:       groupID,
+				SinceUpdateAt: value1.UpdateAt,
+				Cursor: model.PropertyValueSearchCursor{
+					PropertyValueID: value1.ID,
+					CreateAt:        value1.CreateAt,
+				},
+				PerPage: 10,
+			})
+			require.Error(t, err)
 		})
-		require.NoError(t, err)
-		require.Len(t, results, 0)
 	})
 }
 

server/i18n/en.json

@@ -3282,14 +3282,30 @@
     "id": "api.property_field.delete.no_permission.app_error",
     "translation": "You do not have permission to delete this property field."
   },
+  {
+    "id": "api.property_field.get.invalid_opts.app_error",
+    "translation": "Invalid property field search options."
+  },
   {
     "id": "api.property_field.get.invalid_target_type.app_error",
     "translation": "A valid target_type (system, team, or channel) is required."
   },
+  {
+    "id": "api.property_field.get.scope_conflict.app_error",
+    "translation": "channel_id/team_id cannot be combined with target_type/target_id."
+  },
+  {
+    "id": "api.property_field.get.scope_required.app_error",
+    "translation": "A scope parameter (channel_id, team_id, or target_type) is required."
+  },
   {
     "id": "api.property_field.get.target_id_required.app_error",
     "translation": "A target_id is required when querying by channel or team target_type."
   },
+  {
+    "id": "api.property_field.get.target_type_required.app_error",
+    "translation": "A target_type is required when target_id is provided."
+  },
   {
     "id": "api.property_field.invalid_patch.app_error",
     "translation": "Invalid property field patch."
@@ -3310,6 +3326,10 @@
     "id": "api.property_field.update.no_options_permission.app_error",
     "translation": "You do not have permission to manage options for this property field."
   },
+  {
+    "id": "api.property_value.get.invalid_opts.app_error",
+    "translation": "Invalid property value search options."
+  },
   {
     "id": "api.property_value.invalid_object_type.app_error",
     "translation": "The provided object type is not valid."

server/public/model/client4.go

@@ -690,6 +690,10 @@ func (c *Client4) propertyFieldRoute(groupName, objectType, fieldID string) clie
 	return c.propertyFieldsRoute(groupName, objectType).Join(fieldID)
 }
 
+func (c *Client4) propertyFieldsSearchRoute(groupName string) clientRoute {
+	return newClientRoute("properties").Join("groups", groupName, "fields", "search")
+}
+
 func (c *Client4) propertyValuesRoute(groupName, objectType, targetID string) clientRoute {
 	return newClientRoute("properties").Join("groups", groupName, objectType, "values", targetID)
 }
@@ -8165,11 +8169,23 @@ func (c *Client4) GetPropertyFields(ctx context.Context, groupName, objectType s
 	if search.TargetID != "" {
 		values.Set("target_id", search.TargetID)
 	}
-	if search.CursorID != "" && search.CursorCreateAt > 0 {
+	if search.ChannelID != "" {
+		values.Set("channel_id", search.ChannelID)
+	}
+	if search.TeamID != "" {
+		values.Set("team_id", search.TeamID)
+	}
+	if search.SinceUpdateAt != 0 {
+		values.Set("since", strconv.FormatInt(search.SinceUpdateAt, 10))
+	}
+	if search.CursorID != "" {
 		values.Set("cursor_id", search.CursorID)
+	}
+	if search.CursorCreateAt != 0 {
 		values.Set("cursor_create_at", strconv.FormatInt(search.CursorCreateAt, 10))
-	} else if search.CursorID != "" || search.CursorCreateAt > 0 {
-		return nil, nil, errors.New("both cursor_id and cursor_create_at must be provided together")
+	}
+	if search.CursorUpdateAt != 0 {
+		values.Set("cursor_update_at", strconv.FormatInt(search.CursorUpdateAt, 10))
 	}
 	r, err := c.doAPIGetWithQuery(ctx, c.propertyFieldsRoute(groupName, objectType), values, "")
 	if err != nil {
@@ -8179,6 +8195,15 @@ func (c *Client4) GetPropertyFields(ctx context.Context, groupName, objectType s
 	return DecodeJSONFromResponse[[]*PropertyField](r)
 }
 
+func (c *Client4) SearchPropertyFields(ctx context.Context, groupName string, search PropertyFieldSearch) ([]*PropertyField, *Response, error) {
+	r, err := c.doAPIPostJSON(ctx, c.propertyFieldsSearchRoute(groupName), search)
+	if err != nil {
+		return nil, BuildResponse(r), err
+	}
+	defer closeBody(r)
+	return DecodeJSONFromResponse[[]*PropertyField](r)
+}
+
 func (c *Client4) PatchPropertyField(ctx context.Context, groupName, objectType, fieldID string, patch *PropertyFieldPatch) (*PropertyField, *Response, error) {
 	r, err := c.doAPIPatchJSON(ctx, c.propertyFieldRoute(groupName, objectType, fieldID), patch)
 	if err != nil {
@@ -8202,11 +8227,17 @@ func (c *Client4) GetPropertyValues(ctx context.Context, groupName, objectType,
 	if search.PerPage > 0 {
 		values.Set("per_page", strconv.Itoa(search.PerPage))
 	}
-	if search.CursorID != "" && search.CursorCreateAt > 0 {
+	if search.SinceUpdateAt > 0 {
+		values.Set("since", strconv.FormatInt(search.SinceUpdateAt, 10))
+	}
+	if search.CursorID != "" {
 		values.Set("cursor_id", search.CursorID)
+	}
+	if search.CursorCreateAt > 0 {
 		values.Set("cursor_create_at", strconv.FormatInt(search.CursorCreateAt, 10))
-	} else if search.CursorID != "" || search.CursorCreateAt > 0 {
-		return nil, nil, errors.New("both cursor_id and cursor_create_at must be provided together")
+	}
+	if search.CursorUpdateAt > 0 {
+		values.Set("cursor_update_at", strconv.FormatInt(search.CursorUpdateAt, 10))
 	}
 	r, err := c.doAPIGetWithQuery(ctx, c.propertyValuesRoute(groupName, objectType, targetID), values, "")
 	if err != nil {
@@ -8232,11 +8263,17 @@ func (c *Client4) GetSystemPropertyValues(ctx context.Context, groupName string,
 	if search.PerPage > 0 {
 		values.Set("per_page", strconv.Itoa(search.PerPage))
 	}
-	if search.CursorID != "" && search.CursorCreateAt > 0 {
+	if search.SinceUpdateAt > 0 {
+		values.Set("since", strconv.FormatInt(search.SinceUpdateAt, 10))
+	}
+	if search.CursorID != "" {
 		values.Set("cursor_id", search.CursorID)
+	}
+	if search.CursorCreateAt > 0 {
 		values.Set("cursor_create_at", strconv.FormatInt(search.CursorCreateAt, 10))
-	} else if search.CursorID != "" || search.CursorCreateAt > 0 {
-		return nil, nil, errors.New("both cursor_id and cursor_create_at must be provided together")
+	}
+	if search.CursorUpdateAt > 0 {
+		values.Set("cursor_update_at", strconv.FormatInt(search.CursorUpdateAt, 10))
 	}
 	r, err := c.doAPIGetWithQuery(ctx, c.propertySystemValuesRoute(groupName), values, "")
 	if err != nil {

server/public/model/property_field.go

@@ -455,13 +455,26 @@ func IsValidPropertyFieldObjectType(objectType string) bool {
 	return slices.Contains(validPropertyFieldObjectTypes, objectType)
 }
 
+// PropertyFieldSearchCursor carries two alternative pagination keys because
+// field listings serve two different read patterns:
+//
+//   - Directory listings (no since filter) page in creation order using
+//     CreateAt + PropertyFieldID. CreateAt never changes, so the scan is
+//     stable across concurrent patches.
+//   - Delta sync (SinceUpdateAt > 0) pages in update order using UpdateAt +
+//     PropertyFieldID, matching the ORDER BY the store applies in that mode.
+//
+// IsValid requires exactly one of CreateAt or UpdateAt to be positive
+// alongside a valid PropertyFieldID. An empty cursor is also valid and means
+// "start from the beginning".
 type PropertyFieldSearchCursor struct {
 	PropertyFieldID string
 	CreateAt        int64
+	UpdateAt        int64
 }
 
 func (p PropertyFieldSearchCursor) IsEmpty() bool {
-	return p.PropertyFieldID == "" && p.CreateAt == 0
+	return p.PropertyFieldID == "" && p.CreateAt == 0 && p.UpdateAt == 0
 }
 
 func (p PropertyFieldSearchCursor) IsValid() error {
@@ -469,38 +482,110 @@ func (p PropertyFieldSearchCursor) IsValid() error {
 		return nil
 	}
 
-	if p.CreateAt <= 0 {
-		return errors.New("create at cannot be negative or zero")
-	}
-
 	if !IsValidId(p.PropertyFieldID) {
 		return errors.New("property field id is invalid")
 	}
+
+	hasCreate := p.CreateAt > 0
+	hasUpdate := p.UpdateAt > 0
+	if hasCreate == hasUpdate {
+		return errors.New("cursor must have exactly one of create_at or update_at set")
+	}
 	return nil
 }
 
 // PropertyFieldSearch captures the parameters provided by a client for
-// searching property fields
+// searching property fields.
+//
+// Scope is specified one of two ways (mutually exclusive):
+//   - Hierarchical: ChannelID and/or TeamID — returns rows at the named scope
+//     plus every ancestor above it.
+//   - Single-target: TargetType + TargetID — returns rows for exactly one
+//     resource.
+//
+// SinceUpdateAt > 0 switches the endpoint to delta mode: rows are ordered by
+// update_at, tombstones are included, and pagination must use CursorUpdateAt
+// (CursorCreateAt is used in the default directory mode).
 type PropertyFieldSearch struct {
-	TargetType     string `json:"target_type,omitempty"`
-	TargetID       string `json:"target_id,omitempty"`
-	CursorID       string `json:"cursor_id,omitempty"`
-	CursorCreateAt int64  `json:"cursor_create_at,omitempty"`
-	PerPage        int    `json:"per_page"`
+	ObjectTypes    []string `json:"object_types,omitempty"`
+	TargetType     string   `json:"target_type,omitempty"`
+	TargetID       string   `json:"target_id,omitempty"`
+	ChannelID      string   `json:"channel_id,omitempty"`
+	TeamID         string   `json:"team_id,omitempty"`
+	SinceUpdateAt  int64    `json:"since,omitempty"`
+	CursorID       string   `json:"cursor_id,omitempty"`
+	CursorCreateAt int64    `json:"cursor_create_at,omitempty"`
+	CursorUpdateAt int64    `json:"cursor_update_at,omitempty"`
+	PerPage        int      `json:"per_page"`
 }
 
+// PropertyFieldSearchOpts captures the filters accepted by SearchPropertyFields.
+//
+// Invariants enforced by IsValid:
+//   - ObjectType and ObjectTypes are mutually exclusive.
+//   - Every entry in ObjectTypes must be a valid PSAv2 object type.
+//   - ChannelID/TeamID and TargetType/TargetIDs are mutually exclusive scope modes.
+//   - ChannelID requires TeamID (callers must resolve TeamID before search).
+//   - SinceUpdateAt <= 0 means "no filter".
 type PropertyFieldSearchOpts struct {
-	GroupID        string
+	GroupID string
+	// Deprecated: use ObjectTypes instead. Kept for backwards compatibility
+	// with existing callers; mutually exclusive with ObjectTypes.
 	ObjectType     string
+	ObjectTypes    []string
 	TargetType     string
 	TargetIDs      []string
+	ChannelID      string
+	TeamID         string
 	LinkedFieldID  string
-	SinceUpdateAt  int64 // UpdatedAt after which to send the items
+	SinceUpdateAt  int64
 	IncludeDeleted bool
 	Cursor         PropertyFieldSearchCursor
 	PerPage        int
 }
 
+// IsValid runs the cross-field invariants documented on PropertyFieldSearchOpts.
+func (o PropertyFieldSearchOpts) IsValid() error {
+	if o.ObjectType != "" && len(o.ObjectTypes) > 0 {
+		return errors.New("object_type and object_types are mutually exclusive")
+	}
+
+	if o.ObjectType != "" && !IsValidPropertyFieldObjectType(o.ObjectType) {
+		return fmt.Errorf("invalid object_type %q", o.ObjectType)
+	}
+
+	for _, ot := range o.ObjectTypes {
+		if !IsValidPropertyFieldObjectType(ot) {
+			return fmt.Errorf("invalid object_type %q", ot)
+		}
+	}
+
+	scopeByChanTeam := o.ChannelID != "" || o.TeamID != ""
+	scopeByTarget := o.TargetType != "" || len(o.TargetIDs) > 0
+	if scopeByChanTeam && scopeByTarget {
+		return errors.New("channel_id/team_id cannot be combined with target_type/target_id")
+	}
+
+	if err := o.Cursor.IsValid(); err != nil {
+		return err
+	}
+
+	// Cursor key must match the active ordering: delta mode (SinceUpdateAt>0)
+	// pages by UpdateAt; directory mode pages by CreateAt. A mismatch would
+	// silently skip rows because the WHERE clause references the wrong column.
+	if !o.Cursor.IsEmpty() {
+		deltaMode := o.SinceUpdateAt > 0
+		if deltaMode && o.Cursor.UpdateAt == 0 {
+			return errors.New("cursor_update_at required when since is set")
+		}
+		if !deltaMode && o.Cursor.CreateAt == 0 {
+			return errors.New("cursor_create_at required when since is not set")
+		}
+	}
+
+	return nil
+}
+
 func (pf *PropertyField) GetAttr(key string) any {
 	return pf.Attrs[key]
 }

server/public/model/property_field_test.go

@@ -1245,6 +1245,229 @@ func TestPropertyFieldSearchCursor_IsValid(t *testing.T) {
 		}
 		assert.Error(t, cursor.IsValid())
 	})
+
+	t.Run("valid cursor with UpdateAt only", func(t *testing.T) {
+		cursor := PropertyFieldSearchCursor{
+			PropertyFieldID: NewId(),
+			UpdateAt:        GetMillis(),
+		}
+		assert.NoError(t, cursor.IsValid())
+	})
+
+	t.Run("both CreateAt and UpdateAt set is invalid", func(t *testing.T) {
+		cursor := PropertyFieldSearchCursor{
+			PropertyFieldID: NewId(),
+			CreateAt:        1,
+			UpdateAt:        1,
+		}
+		err := cursor.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "exactly one")
+	})
+
+	t.Run("invalid PropertyFieldID with UpdateAt set", func(t *testing.T) {
+		cursor := PropertyFieldSearchCursor{
+			PropertyFieldID: "invalid",
+			UpdateAt:        GetMillis(),
+		}
+		assert.Error(t, cursor.IsValid())
+	})
+
+	t.Run("ID set but neither CreateAt nor UpdateAt is invalid", func(t *testing.T) {
+		// The cursor is non-empty but cannot be paginated against either
+		// ordering — IsValid must reject it rather than fall through to a
+		// silent miss in the store.
+		cursor := PropertyFieldSearchCursor{
+			PropertyFieldID: NewId(),
+		}
+		err := cursor.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "exactly one")
+	})
+
+	t.Run("IsEmpty returns true when UpdateAt is also zero", func(t *testing.T) {
+		cursor := PropertyFieldSearchCursor{}
+		assert.True(t, cursor.IsEmpty())
+	})
+
+	t.Run("IsEmpty returns false when UpdateAt is set", func(t *testing.T) {
+		cursor := PropertyFieldSearchCursor{UpdateAt: 1}
+		assert.False(t, cursor.IsEmpty())
+	})
+}
+
+func TestPropertyFieldSearchOpts_IsValid(t *testing.T) {
+	validID := NewId()
+
+	t.Run("zero-value opts is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("ObjectType alone is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{ObjectType: PropertyFieldObjectTypeChannel}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("ObjectTypes alone is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{ObjectTypes: []string{PropertyFieldObjectTypeChannel, PropertyFieldObjectTypeSystem}}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("ObjectType and ObjectTypes both set is invalid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			ObjectType:  PropertyFieldObjectTypeChannel,
+			ObjectTypes: []string{PropertyFieldObjectTypeChannel},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "mutually exclusive")
+	})
+
+	t.Run("ObjectTypes containing an invalid value mentions that value", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			ObjectTypes: []string{PropertyFieldObjectTypeChannel, "garbage"},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "garbage")
+	})
+
+	t.Run("legacy ObjectType containing an invalid value is rejected", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{ObjectType: "garbage"}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "garbage")
+	})
+
+	t.Run("ChannelID combined with TargetType is invalid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			ChannelID:  validID,
+			TeamID:     NewId(),
+			TargetType: "channel",
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "channel_id/team_id")
+	})
+
+	t.Run("TeamID combined with TargetIDs is invalid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			TeamID:    NewId(),
+			TargetIDs: []string{NewId()},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "channel_id/team_id")
+	})
+
+	t.Run("ChannelID without TeamID is valid (DM/GM scope)", func(t *testing.T) {
+		// DM/GM channels have no parent team. IsValid must not reject
+		// them; the store layer handles the system → channel hierarchy
+		// when TeamID is empty.
+		opts := PropertyFieldSearchOpts{ChannelID: NewId()}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("TeamID alone is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{TeamID: NewId()}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("ChannelID + TeamID is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{ChannelID: NewId(), TeamID: NewId()}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("TargetType + TargetIDs is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			TargetType: string(PropertyFieldTargetLevelChannel),
+			TargetIDs:  []string{NewId(), NewId()},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("SinceUpdateAt negative is valid (treated as no filter)", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{SinceUpdateAt: -1}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("SinceUpdateAt zero is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{SinceUpdateAt: 0}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("SinceUpdateAt positive is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{SinceUpdateAt: 123}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("delegates to Cursor.IsValid - invalid cursor surfaces", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			Cursor: PropertyFieldSearchCursor{
+				PropertyFieldID: NewId(),
+				CreateAt:        1,
+				UpdateAt:        1,
+			},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "exactly one")
+	})
+
+	t.Run("delegates to Cursor.IsValid - empty cursor is fine", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			TeamID: NewId(),
+			Cursor: PropertyFieldSearchCursor{},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("cursor key must match active ordering: delta mode requires UpdateAt", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			SinceUpdateAt: 100,
+			Cursor: PropertyFieldSearchCursor{
+				PropertyFieldID: validID,
+				CreateAt:        1,
+			},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "cursor_update_at")
+	})
+
+	t.Run("cursor key must match active ordering: directory mode requires CreateAt", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			Cursor: PropertyFieldSearchCursor{
+				PropertyFieldID: validID,
+				UpdateAt:        1,
+			},
+		}
+		err := opts.IsValid()
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "cursor_create_at")
+	})
+
+	t.Run("cursor key match: delta + UpdateAt is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			SinceUpdateAt: 100,
+			Cursor: PropertyFieldSearchCursor{
+				PropertyFieldID: validID,
+				UpdateAt:        50,
+			},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("cursor key match: directory + CreateAt is valid", func(t *testing.T) {
+		opts := PropertyFieldSearchOpts{
+			Cursor: PropertyFieldSearchCursor{
+				PropertyFieldID: validID,
+				CreateAt:        50,
+			},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
 }
 
 func TestPluginPropertyOption(t *testing.T) {

server/public/model/property_value.go

@@ -102,13 +102,26 @@ func (pv *PropertyValue) IsValid() error {
 	return nil
 }
 
+// PropertyValueSearchCursor carries two alternative pagination keys because
+// value listings serve two different read patterns:
+//
+//   - Directory listings (no since filter) page in creation order using
+//     CreateAt + PropertyValueID. CreateAt never changes, so the scan is
+//     stable across concurrent updates.
+//   - Delta sync (SinceUpdateAt > 0) pages in update order using UpdateAt +
+//     PropertyValueID, matching the ORDER BY the store applies in that mode.
+//
+// IsValid requires exactly one of CreateAt or UpdateAt to be positive
+// alongside a valid PropertyValueID. An empty cursor is also valid and means
+// "start from the beginning".
 type PropertyValueSearchCursor struct {
 	PropertyValueID string
 	CreateAt        int64
+	UpdateAt        int64
 }
 
 func (p PropertyValueSearchCursor) IsEmpty() bool {
-	return p.PropertyValueID == "" && p.CreateAt == 0
+	return p.PropertyValueID == "" && p.CreateAt == 0 && p.UpdateAt == 0
 }
 
 func (p PropertyValueSearchCursor) IsValid() error {
@@ -116,33 +129,67 @@ func (p PropertyValueSearchCursor) IsValid() error {
 		return nil
 	}
 
-	if p.CreateAt <= 0 {
-		return errors.New("create at cannot be negative or zero")
+	if !IsValidId(p.PropertyValueID) {
+		return errors.New("property value id is invalid")
 	}
 
-	if !IsValidId(p.PropertyValueID) {
-		return errors.New("property field id is invalid")
+	hasCreate := p.CreateAt > 0
+	hasUpdate := p.UpdateAt > 0
+	if hasCreate == hasUpdate {
+		return errors.New("cursor must have exactly one of create_at or update_at set")
 	}
 	return nil
 }
 
+// PropertyValueSearchOpts captures the filters accepted by SearchPropertyValues.
+//
+// SinceUpdateAt > 0 switches the endpoint to delta mode: rows are ordered by
+// UpdateAt, tombstones are included automatically, and pagination must use
+// Cursor.UpdateAt (Cursor.CreateAt is used in the default directory mode).
 type PropertyValueSearchOpts struct {
 	GroupID        string
 	TargetType     string
 	TargetIDs      []string
 	FieldID        string
-	SinceUpdateAt  int64 // UpdateAt after which to send the items
+	SinceUpdateAt  int64
 	IncludeDeleted bool
 	Cursor         PropertyValueSearchCursor
 	PerPage        int
 	Value          json.RawMessage
 }
 
+func (o PropertyValueSearchOpts) IsValid() error {
+	if err := o.Cursor.IsValid(); err != nil {
+		return err
+	}
+
+	// Cursor key must match the active ordering: delta mode (SinceUpdateAt>0)
+	// pages by UpdateAt; directory mode pages by CreateAt. A mismatch would
+	// silently skip rows because the WHERE clause references the wrong column.
+	if !o.Cursor.IsEmpty() {
+		deltaMode := o.SinceUpdateAt > 0
+		if deltaMode && o.Cursor.UpdateAt == 0 {
+			return errors.New("cursor_update_at required when since is set")
+		}
+		if !deltaMode && o.Cursor.CreateAt == 0 {
+			return errors.New("cursor_create_at required when since is not set")
+		}
+	}
+
+	return nil
+}
+
 // PropertyValueSearch captures the parameters provided by a client for
-// searching property values
+// searching property values.
+//
+// SinceUpdateAt > 0 switches the endpoint to delta mode: rows are ordered by
+// update_at, tombstones are included, and pagination must use CursorUpdateAt
+// (CursorCreateAt is used in the default directory mode).
 type PropertyValueSearch struct {
 	CursorID       string `json:"cursor_id,omitempty"`
 	CursorCreateAt int64  `json:"cursor_create_at,omitempty"`
+	CursorUpdateAt int64  `json:"cursor_update_at,omitempty"`
+	SinceUpdateAt  int64  `json:"since,omitempty"`
 	PerPage        int    `json:"per_page"`
 }
 

server/public/model/property_value_test.go

@@ -220,7 +220,7 @@ func TestPropertyValueSearchCursor_IsValid(t *testing.T) {
 		assert.NoError(t, cursor.IsValid())
 	})
 
-	t.Run("valid cursor", func(t *testing.T) {
+	t.Run("valid cursor with CreateAt", func(t *testing.T) {
 		cursor := PropertyValueSearchCursor{
 			PropertyValueID: NewId(),
 			CreateAt:        GetMillis(),
@@ -228,6 +228,14 @@ func TestPropertyValueSearchCursor_IsValid(t *testing.T) {
 		assert.NoError(t, cursor.IsValid())
 	})
 
+	t.Run("valid cursor with UpdateAt", func(t *testing.T) {
+		cursor := PropertyValueSearchCursor{
+			PropertyValueID: NewId(),
+			UpdateAt:        GetMillis(),
+		}
+		assert.NoError(t, cursor.IsValid())
+	})
+
 	t.Run("invalid PropertyValueID", func(t *testing.T) {
 		cursor := PropertyValueSearchCursor{
 			PropertyValueID: "invalid",
@@ -236,10 +244,18 @@ func TestPropertyValueSearchCursor_IsValid(t *testing.T) {
 		assert.Error(t, cursor.IsValid())
 	})
 
-	t.Run("zero CreateAt", func(t *testing.T) {
+	t.Run("neither CreateAt nor UpdateAt set", func(t *testing.T) {
 		cursor := PropertyValueSearchCursor{
 			PropertyValueID: NewId(),
-			CreateAt:        0,
+		}
+		assert.Error(t, cursor.IsValid())
+	})
+
+	t.Run("both CreateAt and UpdateAt set", func(t *testing.T) {
+		cursor := PropertyValueSearchCursor{
+			PropertyValueID: NewId(),
+			CreateAt:        GetMillis(),
+			UpdateAt:        GetMillis(),
 		}
 		assert.Error(t, cursor.IsValid())
 	})
@@ -253,6 +269,79 @@ func TestPropertyValueSearchCursor_IsValid(t *testing.T) {
 	})
 }
 
+func TestPropertyValueSearchOpts_IsValid(t *testing.T) {
+	t.Run("empty opts is valid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("since without cursor is valid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{SinceUpdateAt: 1000}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("delta mode with matching cursor_update_at is valid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			SinceUpdateAt: 1000,
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: NewId(),
+				UpdateAt:        1500,
+			},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("directory mode with matching cursor_create_at is valid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: NewId(),
+				CreateAt:        1500,
+			},
+		}
+		assert.NoError(t, opts.IsValid())
+	})
+
+	t.Run("delta mode with cursor_create_at is invalid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			SinceUpdateAt: 1000,
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: NewId(),
+				CreateAt:        1500,
+			},
+		}
+		assert.Error(t, opts.IsValid())
+	})
+
+	t.Run("directory mode with cursor_update_at is invalid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: NewId(),
+				UpdateAt:        1500,
+			},
+		}
+		assert.Error(t, opts.IsValid())
+	})
+
+	t.Run("invalid cursor surfaces error", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: "invalid",
+				CreateAt:        1500,
+			},
+		}
+		assert.Error(t, opts.IsValid())
+	})
+
+	t.Run("cursor with ID but neither create_at nor update_at is invalid", func(t *testing.T) {
+		opts := PropertyValueSearchOpts{
+			Cursor: PropertyValueSearchCursor{
+				PropertyValueID: NewId(),
+			},
+		}
+		assert.Error(t, opts.IsValid())
+	})
+}
+
 func TestSanitizePropertyValue(t *testing.T) {
 	cases := []struct {
 		name string