Merges pull request #13 v3.9.2

user manager filter and new timestamps, user profiling info for UI

14 files changed, 628 insertions, 45 deletions

diff --git a/generalresearch/managers/base.py b/generalresearch/managers/base.py
index 257032f..faffddf 100644
--- a/generalresearch/managers/base.py
+++ b/generalresearch/managers/base.py
@@ -21,15 +21,16 @@ class Permission(int, Enum):
 class Manager:
     DEFAULT_PAGE_SIZE = 50
 
+    @classmethod
     def validate_pagination(
-        self,
+        cls,
         page: int | None = 1,
         size: int | None = None,
     ):
         page = page if page is not None else 1
         assert type(page) is int
         assert page >= 1, "page starts at 1"
-        size = size if size is not None else self.DEFAULT_PAGE_SIZE
+        size = size if size is not None else cls.DEFAULT_PAGE_SIZE
         assert type(size) is int
         assert 1 <= size <= 100
         offset = (page - 1) * size
diff --git a/generalresearch/managers/thl/profiling/question.py b/generalresearch/managers/thl/profiling/question.py
index 078894a..7fdd1c2 100644
--- a/generalresearch/managers/thl/profiling/question.py
+++ b/generalresearch/managers/thl/profiling/question.py
@@ -17,7 +17,6 @@ from generalresearch.models.thl.profiling.upk_question import (
 
 
 class QuestionManager(PostgresManager):
-
     def get_multi_upk(self, question_ids: Collection[str]) -> list[UpkQuestion]:
         query = """
         SELECT data, property_code, explanation_template, explanation_fragment_template
@@ -78,7 +77,7 @@ class QuestionManager(PostgresManager):
         return res
 
     @cached(
-        cache=TTLCache(maxsize=256, ttl=3600 + random.randint(-900, 900)),
+        cache=TTLCache(maxsize=1024, ttl=3600 + random.randint(-900, 900)),
         lock=threading.Lock(),
         info=True,
     )
@@ -110,6 +109,39 @@ class QuestionManager(PostgresManager):
         x["data"].pop("categories", None)
         return UpkQuestion.model_validate(x["data"])
 
+    def filter_by_properties_and_locale(
+        self, property_codes: Collection[str], country_iso: str, language_iso: str
+    ) -> list[UpkQuestion]:
+        query = """
+        SELECT data, property_code, explanation_template, explanation_fragment_template
+        FROM marketplace_question
+        WHERE property_code = ANY(%(property_codes)s)
+            AND country_iso = %(country_iso)s
+            AND language_iso = %(language_iso)s
+        """
+        res: list[dict[str, Any]] = self.pg_config.execute_sql_query(
+            query=query,
+            params={
+                "property_codes": list(property_codes),
+                "country_iso": country_iso,
+                "language_iso": language_iso,
+            },
+        )
+        for x in res:
+            x["data"]["ext_question_id"] = x["property_code"]
+            x["data"]["explanation_template"] = x["explanation_template"]
+            x["data"]["explanation_fragment_template"] = x[
+                "explanation_fragment_template"
+            ]
+            x["data"].pop("categories", None)
+        res2 = []
+        for x in res:
+            try:
+                res2.append(UpkQuestion.model_validate(x["data"]))
+            except ValidationError as e:
+                LOG.warning(e)
+        return res2
+
     def filter_by_property(
         self, lookup: Collection[tuple[str, str, str]]
     ) -> list[UpkQuestion]:
diff --git a/generalresearch/managers/thl/profiling/user_profiling_data.py b/generalresearch/managers/thl/profiling/user_profiling_data.py
new file mode 100644
index 0000000..1d29328
--- /dev/null
+++ b/generalresearch/managers/thl/profiling/user_profiling_data.py
@@ -0,0 +1,102 @@
+from typing import Any
+
+from generalresearch.managers.thl.profiling.question import QuestionManager
+from generalresearch.managers.thl.profiling.uqa import UQAManager
+from generalresearch.managers.thl.profiling.user_upk import UserUpkManager
+from generalresearch.managers.thl.userhealth import UserIpHistoryManager
+from generalresearch.models.thl.profiling.upk_property import PropertyType
+from generalresearch.models.thl.profiling.user_info import (
+    MarketProfileKnowledge,
+    UserProfileKnowledge,
+    UserProfilingData,
+)
+from generalresearch.models.thl.profiling.user_question_answer import (
+    MarketplaceResearchProfileQuestion,
+)
+from generalresearch.models.thl.user import User
+from generalresearch.models.thl.user_ref import UserRef
+
+
+def get_user_mrpq(
+    user: User,
+    uqa_manager: UQAManager,
+    question_manager: QuestionManager,
+    country_iso: str | None = None,
+) -> list[MarketplaceResearchProfileQuestion]:
+    res = uqa_manager.get(user=user)
+    if country_iso:
+        res = [x for x in res if x.country_iso == country_iso]
+    mrpqs = []
+    for x in res:
+        mrpqs.extend(x.get_mrpqs(question_manager=question_manager))
+    return mrpqs
+
+
+def get_user_upk_formatted(
+    user_id: int, user_upk_manager: UserUpkManager, country_iso: str | None = None
+):
+    res = user_upk_manager.get_user_upk(user_id=user_id)
+    if country_iso:
+        res = [x for x in res if x.country_iso == country_iso]
+    new_res = []
+    for x in res:
+        d: dict[str, Any] = {
+            "property_id": x.property_id,
+            "property_label": x.property_label,
+            "created": x.created,
+            "translation": "",
+        }
+        if x.prop_type in {PropertyType.UPK_NUMERICAL, PropertyType.UPK_TEXT}:
+            d["answer"] = [{"value": str(x.value)}]
+        else:
+            d["answer"] = [{"id": x.item_id, "label": x.item_label, "translation": ""}]
+        new_res.append(d)
+    d = {}
+    for x in new_res:
+        if x["property_id"] in d:
+            d[x["property_id"]]["answer"].extend(x["answer"])
+        else:
+            d[x["property_id"]] = x
+    return list(d.values())
+
+
+def get_user_profiling_data(
+    user_upk_manager: UserUpkManager,
+    uqa_manager: UQAManager,
+    question_manager: QuestionManager,
+    user: User,
+    country_iso: str | None = None,
+    user_ip_history_manager: UserIpHistoryManager | None = None,
+) -> UserProfilingData:
+    """
+    Get user's mrpq and upk data, for API response
+    """
+    assert user.user_id is not None
+    user = user if isinstance(user, UserRef) else user.to_user_ref()
+
+    if country_iso is None and user_ip_history_manager is not None:
+        country_iso = user_ip_history_manager.get_user_latest_country(user)
+
+    mrpq = get_user_mrpq(
+        user=user,
+        uqa_manager=uqa_manager,
+        question_manager=question_manager,
+        country_iso=country_iso,
+    )
+    mrpq = [
+        MarketProfileKnowledge.from_MarketplaceResearchProfileQuestion(x) for x in mrpq
+    ]
+
+    upk = get_user_upk_formatted(
+        user_id=user.user_id,
+        user_upk_manager=user_upk_manager,
+        country_iso=country_iso,
+    )
+    upk = [UserProfileKnowledge.model_validate(x) for x in upk]
+
+    d = {
+        "user_profile_knowledge": upk,
+        "marketplace_profile_knowledge": mrpq,
+        "user": user,
+    }
+    return UserProfilingData.model_validate(d)
diff --git a/generalresearch/managers/thl/user_manager/mysql_user_manager.py b/generalresearch/managers/thl/user_manager/mysql_user_manager.py
index d2732ed..82e504c 100644
--- a/generalresearch/managers/thl/user_manager/mysql_user_manager.py
+++ b/generalresearch/managers/thl/user_manager/mysql_user_manager.py
@@ -4,15 +4,18 @@ import logging
 import operator
 from collections.abc import Collection
 from datetime import UTC, datetime
+from enum import StrEnum
 from threading import Lock
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 from uuid import uuid4
 
 import psycopg
 from cachetools import LRUCache, cachedmethod
-from psycopg import sql
+from psycopg import Connection, sql
+from pydantic import NonNegativeInt
 
-from generalresearch.models.custom_types import UUIDStr
+from generalresearch.managers.base import Permission, PostgresManager
+from generalresearch.models.custom_types import UUIDStr, try_normalize_email
 from generalresearch.models.thl.user import User
 
 if TYPE_CHECKING:
@@ -23,25 +26,79 @@ logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
 
-class MysqlUserManager:
-    def __init__(self, pg_config: PostgresConfig, is_read_replica: bool):
-        self.pg_config = pg_config
+class LastActivityColumn(StrEnum):
+    LAST_SEEN = "last_seen"
+    LAST_TASK_COMPLETED_AT = "last_task_completed_at"
+    LAST_OFFERWALL_REQUESTED_AT = "last_offerwall_requested_at"
+    LAST_PROFILING_SUBMISSION_AT = "last_profiling_submission_at"
+
+
+class UserOrderField(StrEnum):
+    CREATED = "created"
+    LAST_SEEN = "last_seen"
+    LAST_TASK_COMPLETED_AT = "last_task_completed_at"
+    LAST_OFFERWALL_REQUESTED_AT = "last_offerwall_requested_at"
+    LAST_PROFILING_SUBMISSION_AT = "last_profiling_submission_at"
+    PRODUCT_USER_ID = "product_user_id"
+
+
+class MysqlUserManager(PostgresManager):
+    def __init__(
+        self,
+        pg_config: PostgresConfig,
+        is_read_replica: bool,
+        permissions: Collection[Permission] | None = None,
+    ):
         self.is_read_replica = is_read_replica
         self.product_id_exists_cache = LRUCache(maxsize=5000)
         self.product_id_exists_cache_lock = Lock()
+        super().__init__(pg_config=pg_config, permissions=permissions)
 
-    def _set_last_seen(self, user: User) -> None:
-        # Don't call this directly. Use UserManager.set_last_seen()
+    def _set_last_activity(
+        self,
+        user: User,
+        column: LastActivityColumn,
+        conn: Connection | None = None,
+    ) -> None:
+        # Don't call this directly. Use UserManager.set_last_activity().
         assert not self.is_read_replica
-        now = datetime.now(tz=UTC)
-        self.pg_config.execute_write(
-            """
+        column = LastActivityColumn(column)
+
+        with self.connection(conn) as conn, conn.cursor() as c:
+            c.execute(
+                sql.SQL("UPDATE thl_user SET {} = %s WHERE id = %s").format(
+                    sql.Identifier(column)
+                ),
+                params=[getattr(user, column), user.user_id],
+            )
+
+    def _set_last_country_ip(
+        self,
+        user: User,
+        last_ip: str,
+        last_country_iso: str | None = None,
+        conn: Connection | None = None,
+    ) -> None:
+        # Don't call this directly. Use UserManager.set_last_country_ip()
+        assert not self.is_read_replica
+        params = {
+            "user_id": user.user_id,
+            "last_ip": last_ip,
+            "last_country_iso": last_country_iso,
+        }
+        with self.connection(conn) as conn, conn.cursor() as c:
+            c.execute(
+                """
             UPDATE thl_user
-            SET last_seen = %s
-            WHERE id = %s
-            """,
-            params=[now, user.user_id],
-        )
+            SET last_ip = %(last_ip)s, last_country_iso = %(last_country_iso)s
+            WHERE id = %(user_id)s
+            AND (
+                last_ip IS DISTINCT FROM %(last_ip)s OR
+                last_country_iso IS DISTINCT FROM %(last_country_iso)s
+            )
+                """,
+                params,
+            )
 
     def _change_product_user_id(self, *, user: User, new_product_user_id: str) -> User:
         """Change a user's supplier-provided ID in the primary database."""
@@ -168,7 +225,6 @@ class MysqlUserManager:
             "last_seen": now,
         }
 
-        # in postgres, you do not include the auto-increment id column
         query = sql.SQL("""
         INSERT INTO thl_user
         (uuid, product_id, product_user_id, created, 
@@ -182,6 +238,15 @@ class MysqlUserManager:
             with self.pg_config.make_connection() as conn, conn.cursor() as c:
                 c.execute(query=query, params=params)
                 user_id = c.fetchone()["id"]
+                c.execute(
+                    """
+                    UPDATE userprofile_brokerageproduct
+                    SET user_count = user_count + 1
+                    WHERE id = %(product_id)s
+                    """,
+                    params=params,
+                )
+                assert c.rowcount == 1
         except psycopg.IntegrityError:
             # Two machines/processes are trying to create this same (product_id, product_user_id)
             #   at the same time. There's a unique index, so mysql will not let two be created.
@@ -327,3 +392,186 @@ class MysqlUserManager:
                 params={"user_uuids": user_uuids},
             )
         return [User.from_db(x) for x in res]
+
+    @staticmethod
+    def make_filter_str(
+        product_id: str | None = None,
+        business_id: str | None = None,
+        team_id: str | None = None,
+        product_user_id: str | None = None,
+        email_startswith: str | None = None,
+        display_name_like: str | None = None,
+        allow_global_search: bool = False,
+    ) -> tuple[str, dict[str, Any]]:
+        filters = []
+        params = {}
+
+        if product_id is not None:
+            params["product_id"] = product_id
+            filters.append("product_id = %(product_id)s")
+        if business_id:
+            params["business_id"] = business_id
+            filters.append("ubp.business_id = %(business_id)s")
+        if team_id:
+            params["team_id"] = team_id
+            filters.append("ubp.team_id = %(team_id)s")
+        if product_user_id:
+            # This should only be allowed within the context of a product, team, or business,
+            # unless we are admins
+            if not allow_global_search:
+                assert any([product_id, business_id, team_id]), (
+                    "must scope to a product, team, or business"
+                )
+            params["product_user_id"] = product_user_id
+            filters.append("u.product_user_id = %(product_user_id)s")
+        if email_startswith is not None:
+            if not allow_global_search:
+                assert any([product_id, business_id, team_id]), (
+                    "must scope to a product, team, or business"
+                )
+            params["email_startswith"] = f"{email_startswith}%"
+            filters.append("email_address LIKE %(email_startswith)s")
+            normalized_email = try_normalize_email(email_startswith)
+            if normalized_email != email_startswith:
+                params["normalized_email_startswith"] = f"{normalized_email}%"
+                filters.append("canonical_email LIKE %(normalized_email_startswith)s")
+        if display_name_like is not None:
+            if not allow_global_search:
+                assert any([product_id, business_id, team_id]), (
+                    "must scope to a product, team, or business"
+                )
+            params["display_name_like"] = f"{display_name_like}%"
+            filters.append("display_name ILIKE %(display_name_like)s")
+
+        return "WHERE " + " AND ".join(filters) if filters else "", params
+
+    def filter(
+        self,
+        product_id: str | None = None,
+        business_id: str | None = None,
+        team_id: str | None = None,
+        product_user_id: str | None = None,
+        email_startswith: str | None = None,
+        display_name_like: str | None = None,
+        page: int | None = None,
+        size: int | None = None,
+        order_field: UserOrderField = UserOrderField.CREATED,
+        descending: bool = False,
+        allow_global_search: bool = False,
+    ) -> tuple[list[User], NonNegativeInt]:
+        with self.pg_config.make_connection() as conn:
+            items = self.filter_page(
+                product_id=product_id,
+                business_id=business_id,
+                team_id=team_id,
+                product_user_id=product_user_id,
+                email_startswith=email_startswith,
+                display_name_like=display_name_like,
+                page=page,
+                size=size,
+                order_field=order_field,
+                descending=descending,
+                allow_global_search=allow_global_search,
+                conn=conn,
+            )
+            total = self.filter_count(
+                product_id=product_id,
+                business_id=business_id,
+                team_id=team_id,
+                product_user_id=product_user_id,
+                email_startswith=email_startswith,
+                display_name_like=display_name_like,
+                allow_global_search=allow_global_search,
+                conn=conn,
+            )
+        return items, total
+
+    def filter_page(
+        self,
+        product_id: str | None = None,
+        business_id: str | None = None,
+        team_id: str | None = None,
+        product_user_id: str | None = None,
+        email_startswith: str | None = None,
+        display_name_like: str | None = None,
+        page: int | None = None,
+        size: int | None = None,
+        order_field: UserOrderField = UserOrderField.CREATED,
+        descending: bool = False,
+        allow_global_search: bool = False,
+        conn: Connection | None = None,
+    ) -> list[User]:
+        pagination_sql = self.validate_pagination(page=page, size=size)
+        filter_str, params = self.make_filter_str(
+            product_id=product_id,
+            business_id=business_id,
+            team_id=team_id,
+            product_user_id=product_user_id,
+            email_startswith=email_startswith,
+            display_name_like=display_name_like,
+            allow_global_search=allow_global_search,
+        )
+        order_field = UserOrderField(order_field)
+        order_by_sql = (
+            f"u.{order_field.value} {'DESC' if descending else 'ASC'}, u.uuid"
+        )
+
+        query = f"""
+        SELECT u.id as user_id, u.uuid, u.product_id, u.product_user_id, u.blocked,
+            u.created, u.last_seen, u.last_task_completed_at,
+            u.last_offerwall_requested_at, u.last_profiling_submission_at,
+            tu.email_address, tu.display_name, ubp.business_id, ubp.team_id
+        FROM thl_user u
+        LEFT JOIN thl_usermetadata tu on u.id = tu.user_id
+        JOIN userprofile_brokerageproduct ubp ON ubp.id = u.product_id
+        {filter_str}
+        ORDER BY {order_by_sql}
+        {pagination_sql}
+        """
+        with self.connection(conn) as conn, conn.cursor() as c:
+            c.execute(query, params)
+            res = c.fetchall()
+
+        return [User.from_db(x) for x in res]
+
+    def filter_count(
+        self,
+        product_id: str | None = None,
+        business_id: str | None = None,
+        team_id: str | None = None,
+        product_user_id: str | None = None,
+        email_startswith: str | None = None,
+        display_name_like: str | None = None,
+        allow_global_search: bool = False,
+        conn: Connection | None = None,
+    ) -> NonNegativeInt:
+
+        filter_str, params = self.make_filter_str(
+            product_id=product_id,
+            business_id=business_id,
+            team_id=team_id,
+            product_user_id=product_user_id,
+            email_startswith=email_startswith,
+            display_name_like=display_name_like,
+            allow_global_search=allow_global_search,
+        )
+        if not any([product_user_id, email_startswith, display_name_like]):
+            # We're filtering on product, team, or business only. So use the
+            # precomputed counts
+            query = f"""
+            SELECT SUM(user_count) AS cnt
+            FROM userprofile_brokerageproduct ubp
+                {filter_str}
+            """
+        else:
+            query = f"""
+            SELECT COUNT(1) AS cnt
+            FROM thl_user u
+            LEFT JOIN thl_usermetadata tu on u.id = tu.user_id
+            JOIN userprofile_brokerageproduct ubp ON ubp.id = u.product_id
+                {filter_str}
+            """
+        with self.connection(conn) as conn, conn.cursor() as c:
+            c.execute(query, params)
+            res = c.fetchall()
+        return res[0]["cnt"]
diff --git a/generalresearch/managers/thl/user_manager/user_manager.py b/generalresearch/managers/thl/user_manager/user_manager.py
index f322d2b..12ef474 100644
--- a/generalresearch/managers/thl/user_manager/user_manager.py
+++ b/generalresearch/managers/thl/user_manager/user_manager.py
@@ -3,11 +3,12 @@ from __future__ import annotations
 import logging
 import operator
 from collections.abc import Collection
-from datetime import datetime
+from datetime import UTC, datetime
 from threading import Lock
 from typing import TYPE_CHECKING
 
 from cachetools import cachedmethod
+from cachetools.keys import hashkey
 from pydantic import RedisDsn
 
 from generalresearch.cacheing import InstrumentedTTLCache
@@ -17,6 +18,7 @@ from generalresearch.managers.thl.user_manager.exceptions import (
     UserDoesntExistError,
 )
 from generalresearch.managers.thl.user_manager.mysql_user_manager import (
+    LastActivityColumn,
     MysqlUserManager,
 )
 from generalresearch.managers.thl.user_manager.rate_limit import (
@@ -92,8 +94,27 @@ class UserManager:
         self.get_user_cache_lock = Lock()
 
     def set_last_seen(self, user: User) -> None:
+        return self.set_last_activity(user, LastActivityColumn.LAST_SEEN)
+
+    def set_last_activity(self, user: User, column: LastActivityColumn) -> None:
+        assert Permission.UPDATE in self.sql_permissions, "permission error"
+        assert self.redis_user_manager is not None
+        assert user.user_id is not None
+        column = LastActivityColumn(column)
+        key = f"user-activity:{user.user_id}:{column.value}"
+        if self.redis_user_manager.client.set(key, "1", nx=True, ex=60):
+            setattr(user, column, datetime.now(tz=UTC))
+            self.mysql_user_manager._set_last_activity(user, column)
+            self.cache_clear_user(user)
+            self.set_user_inmemory_cache(user)
+
+    def set_last_country_ip(
+        self, user: User, last_ip: str, last_country_iso: str | None = None
+    ) -> None:
         assert Permission.UPDATE in self.sql_permissions, "permission error"
-        return self.mysql_user_manager._set_last_seen(user)
+        return self.mysql_user_manager._set_last_country_ip(
+            user, last_ip, last_country_iso
+        )
 
     def change_product_user_id(self, *, user: User, new_product_user_id: str) -> User:
         """Change a user's supplier-provided ID and refresh user lookup caches.
@@ -119,7 +140,8 @@ class UserManager:
             user=user, new_product_user_id=new_product_user_id
         )
 
-        self.cache_clear()
+        self.cache_clear_user(user)
+        self.cache_clear_user(updated_user)
         if self.redis_user_manager:
             self.redis_user_manager.clear_user(user)
         return updated_user
@@ -150,6 +172,20 @@ class UserManager:
         with self.get_user_cache_lock:
             self.get_user_cache.clear()
 
+    def cache_clear_user(self, user: User) -> None:
+        """Remove every local ``get_user`` cache entry for one user."""
+        keys = (
+            hashkey(user_id=user.user_id),
+            hashkey(user_uuid=user.uuid),
+            hashkey(
+                product_id=user.product_id,
+                product_user_id=user.product_user_id,
+            ),
+        )
+        with self.get_user_cache_lock:
+            for key in keys:
+                self.get_user_cache.pop(key, None)
+
     def cache_info(self):
         with self.get_user_cache_lock:
             return self.get_user_cache.cache_info()
@@ -369,7 +405,7 @@ class UserManager:
         # If we change something about a user, we should update the in-memory caches
         self.set_user_inmemory_cache(user)
         # Clear local cached copies so the blocked state is returned immediately.
-        self.cache_clear()
+        self.cache_clear_user(user)
         return True
 
     def is_whitelisted(self, user: User) -> bool:
diff --git a/generalresearch/models/custom_types.py b/generalresearch/models/custom_types.py
index 0040b92..26c8b7e 100644
--- a/generalresearch/models/custom_types.py
+++ b/generalresearch/models/custom_types.py
@@ -47,6 +47,11 @@ def validate_canonical_email(value: str) -> str:
         )
     return value
 
+def try_normalize_email(value: str) -> str:
+    try:
+        return email_normalize.normalize(value).normalized_address
+    except ValueError:
+        return value
 
 CanonicalEmailStr = Annotated[
     EmailStr,
diff --git a/generalresearch/models/thl/profiling/user_info.py b/generalresearch/models/thl/profiling/user_info.py
index 46733d4..34c957b 100644
--- a/generalresearch/models/thl/profiling/user_info.py
+++ b/generalresearch/models/thl/profiling/user_info.py
@@ -1,14 +1,13 @@
 from __future__ import annotations
 
 from pydantic import BaseModel, ConfigDict, Field
-from pydantic.json_schema import SkipJsonSchema
 
 from generalresearch.models.custom_types import AwareDatetimeISO
 from generalresearch.models.definitions import Source
 from generalresearch.models.thl.profiling.user_question_answer import (
     MarketplaceResearchProfileQuestion,
 )
-from generalresearch.models.thl.user import User
+from generalresearch.models.thl.user_ref import UserRef
 
 
 class UserProfileKnowledgeAnswer(BaseModel):
@@ -43,14 +42,20 @@ class MarketProfileKnowledge(BaseModel):
     )
 
     question_id: str = Field(examples=["gender", "1843", "gender_plus"])
+    question_text: str | None = Field(default=None)
 
     answer: list[str] = Field(
         default_factory=list, examples=[["male"], ["7657644"], ["1"]]
     )
+    answer_text: list[str] | None = Field(default=None)
 
     created: AwareDatetimeISO = Field(
         description="When the User submitted this Profiling data"
     )
+    is_derived: bool = Field(
+        default=False,
+        description='What this answer derived from a "harmonized" / "gold-standard" question?',
+    )
 
     @classmethod
     def from_MarketplaceResearchProfileQuestion(
@@ -61,16 +66,24 @@ class MarketProfileKnowledge(BaseModel):
             question_id=q.question_code,
             answer=list(q.answer),
             created=q.timestamp,
+            question_text=q.question_text,
+            answer_text=list(q.answer_text) if q.answer_text is not None else None,
+            is_derived=q.is_derived,
         )
 
 
-class UserInfo(BaseModel):
+class UserProfilingData(BaseModel):
     model_config = ConfigDict()
 
-    user: SkipJsonSchema[User | None] = Field(exclude=True, default=None)
+    user: UserRef = Field()
 
     user_profile_knowledge: list[UserProfileKnowledge] = Field(default_factory=list)
 
     marketplace_profile_knowledge: list[MarketProfileKnowledge] = Field(
         default_factory=list
     )
+
+
+class UserInfo(UserProfilingData):
+    # fsb endpoint uses this name. todo: rename it
+    pass
diff --git a/generalresearch/models/thl/profiling/user_question_answer.py b/generalresearch/models/thl/profiling/user_question_answer.py
index 342b65e..bef844b 100644
--- a/generalresearch/models/thl/profiling/user_question_answer.py
+++ b/generalresearch/models/thl/profiling/user_question_answer.py
@@ -14,6 +14,7 @@ from pydantic import (
     model_validator,
 )
 
+from generalresearch.managers.thl.profiling.question import QuestionManager
 from generalresearch.models.custom_types import AwareDatetimeISO, UUIDStr
 from generalresearch.models.definitions import MAX_INT32, Source
 from generalresearch.models.thl.locales import CountryISO, LanguageISO
@@ -57,6 +58,10 @@ class UserQuestionAnswer(BaseModel):
         )
         return calc_answers
 
+    @property
+    def is_gold_standard(self) -> bool:
+        return self.property_code.startswith("gr:")
+
     def model_dump_mysql(self, session_id: str | None = None) -> dict[str, Any]:
         d = self.model_dump(mode="json", exclude={"calc_answers", "timestamp"})
         d["answer"] = json.dumps(self.answer)
@@ -67,17 +72,36 @@ class UserQuestionAnswer(BaseModel):
         d["session_id"] = session_id
         return d
 
-    def get_mrpqs(self) -> Iterator[MarketplaceResearchProfileQuestion]:
+    def get_mrpqs(
+        self, question_manager: QuestionManager
+    ) -> Iterator[MarketplaceResearchProfileQuestion]:
         assert self.calc_answers
+        property_codes = set(self.calc_answers.keys())
+        qs = question_manager.filter_by_properties_and_locale(
+            property_codes, country_iso=self.country_iso, language_iso=self.language_iso
+        )
+        text_lookup = {q.ext_question_id: q.text for q in qs}
+        ans_lookup = {
+            q.ext_question_id: {c.id: c.text for c in q.choices}
+            for q in qs
+            if q.choices is not None
+        }
+
         for k, v in self.calc_answers.items():
             source, question_code = k.split(":", 1)
+            ans = ans_lookup.get(k, {})
+            ans_text = tuple(sorted([ans.get(vv, "") for vv in v]))
+            ans_text = ans_text or None
             yield MarketplaceResearchProfileQuestion(
                 question_code=question_code,
-                source=source,
+                question_text=text_lookup.get(k),
+                source=Source(source),
                 country_iso=self.country_iso,
                 language_iso=self.language_iso,
                 answer=tuple(sorted(set(v))),
+                answer_text=ans_text,
                 timestamp=self.timestamp,
+                is_derived=self.is_gold_standard
             )
 
     @field_validator("answer")
@@ -124,17 +148,23 @@ DUMMY_UQA = UserQuestionAnswer(
 
 
 class MarketplaceResearchProfileQuestion(BaseModel):
-    """Answer submitted to a question by a user, that has been transformed
+    """Answer submitted to a question by a user that has been transformed
     into a question answer that is specific to a marketplace."""
 
     question_code: str = Field(
         description="# the question id/code on the marketplace", min_length=1
     )
+    question_text: str | None = Field(default=None)
     source: Source = Field()  # the one or two-letter marketplace code
     answer: tuple[str, ...] = Field(min_length=1)
+    answer_text: tuple[str, ...] | None = Field(min_length=1, default=None)
     timestamp: AwareDatetimeISO = Field()
     country_iso: CountryISO = Field()
     language_iso: LanguageISO = Field()
+    is_derived: bool = Field(
+        default=False,
+        description='What this answer derived from a "harmonized" / "gold-standard" question?',
+    )
 
     @model_validator(mode="after")
     def validate_keys(self):
diff --git a/generalresearch/models/thl/user.py b/generalresearch/models/thl/user.py
index 59a14db..ba49104 100644
--- a/generalresearch/models/thl/user.py
+++ b/generalresearch/models/thl/user.py
@@ -77,11 +77,23 @@ class User(BaseModel):
         description="When the user was created on the GRL platform.",
     )
 
-    # Note: due to cacheing, last_seen might be up to a day out of date!
+    # Note: due to cacheing, last_* timestamps might be up to a day out of date!
+    # Updates are rate-limited to once per minute
     last_seen: AwareDatetimeISO | None = Field(
         default=None,
-        description="When the user was last seen on, or acting on any"
-        "part of the GRL platform.",
+        description="When the user last started a task or returned from a marketplace redirect.",
+    )
+    last_task_completed_at: AwareDatetimeISO | None = Field(
+        default=None,
+        description="When the user most recently completed a task.",
+    )
+    last_offerwall_requested_at: AwareDatetimeISO | None = Field(
+        default=None,
+        description="When the user most recently requested an offerwall.",
+    )
+    last_profiling_submission_at: AwareDatetimeISO | None = Field(
+        default=None,
+        description="When the user most recently submitted profiling answers.",
     )
 
     # --- Prefetch Fields ---
diff --git a/generalresearch/thl_django/common/models.py b/generalresearch/thl_django/common/models.py
index 67c5ca5..968171f 100644
--- a/generalresearch/thl_django/common/models.py
+++ b/generalresearch/thl_django/common/models.py
@@ -2,7 +2,7 @@ import uuid
 
 from django.db import models
 from django.db.models import Q
-
+from django.contrib.postgres.indexes import GinIndex
 
 class THLSession(models.Model):
     """
@@ -305,14 +305,25 @@ class THLUser(models.Model):
     # This will be useful for looking at signups per country globally.
     created = models.DateTimeField(null=False)
 
-    # This will be useful as our Daily Active User metric (DAU) that a lot
-    # of marketplaces want reported on. We'll be able to change the
-    # logic for when this is updated, but keeping it in a table will
-    # make queries much easier.
-    # We'll force null=True, and his will be the same as created
-    # until they're "seen again".
+    # The most recent time the user started a task or returned from a marketplace
+    # redirect. This is initialized to `created` and can be used to report users
+    # who were recently active in tasks.
     last_seen = models.DateTimeField(null=False)
 
+    # The time the user most recently completed a task.
+    last_task_completed_at = models.DateTimeField(null=True)
+
+    # The time of the user's most recent offerwall request. Database writes are
+    # rate-limited through Redis to at most once per minute to reduce table churn,
+    # so this value may lag the latest request by up to one minute.
+    last_offerwall_requested_at = models.DateTimeField(null=True)
+
+    # The time the user most recently submitted profiling answers, either during
+    # a GRS session or directly through the API. Database writes are rate-limited
+    # through Redis to at most once per minute, so this value may lag the latest
+    # submission by up to one minute.
+    last_profiling_submission_at = models.DateTimeField(null=True)
+
     # this isn't used for any security measures, but we have an increasing
     # need to provide a "live panel book" to describe our user base. This
     # will help immediately filter users down by "after users from X country
@@ -352,6 +363,22 @@ class THLUser(models.Model):
             # models.Index(fields=["product_id", "product_user_id"]),
             models.Index(fields=["created"]),
             models.Index(fields=["last_seen"]),
+            models.Index(fields=["product_id", "-last_seen"]),
+            models.Index(
+                fields=["last_task_completed_at"],
+                name="thl_user_last_task_comp_nn",
+                condition=Q(last_task_completed_at__isnull=False),
+            ),
+            models.Index(
+                fields=["last_offerwall_requested_at"],
+                name="thl_user_last_offerwall_nn",
+                condition=Q(last_offerwall_requested_at__isnull=False),
+            ),
+            models.Index(
+                fields=["last_profiling_submission_at"],
+                name="thl_user_last_profile_nn",
+                condition=Q(last_profiling_submission_at__isnull=False),
+            ),
             models.Index(fields=["last_country_iso"]),
         ]
 
@@ -379,8 +406,7 @@ class THLUserMetadata(models.Model):
     canonical_email = models.CharField(max_length=320, null=True)
 
     # Not unique within a BP, or anything like that. A user
-    #   can set this to whatever they like. No index
-    #   as we will not ever look up a user by their name.
+    #   can set this to whatever they like.
     display_name = models.CharField(max_length=255, null=True)
 
     class Meta:
@@ -391,6 +417,11 @@ class THLUserMetadata(models.Model):
             models.Index(fields=["email_sha1"]),
             models.Index(fields=["email_md5"]),
             models.Index(fields=["canonical_email"]),
+            GinIndex(
+                fields=["display_name"],
+                opclasses=["gin_trgm_ops"],
+                name="usermetadata_name_trgm",
+            ),
         ]
 
 
diff --git a/generalresearch/thl_django/migrations/0014_brokerageproduct_user_count_and_more.py b/generalresearch/thl_django/migrations/0014_brokerageproduct_user_count_and_more.py
new file mode 100644
index 0000000..e6a5c96
--- /dev/null
+++ b/generalresearch/thl_django/migrations/0014_brokerageproduct_user_count_and_more.py
@@ -0,0 +1,24 @@
+# Generated by Django 6.1.1 on 2026-10-23 19:13
+
+import django.contrib.postgres.indexes
+from django.db import migrations, models
+from django.contrib.postgres.operations import TrigramExtension
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('thl_django', '0013_brokerageproduct_balance_and_more'),
+    ]
+
+    operations = [
+        TrigramExtension(),
+        migrations.AddField(
+            model_name='brokerageproduct',
+            name='user_count',
+            field=models.IntegerField(default=0),
+        ),
+        migrations.AddIndex(
+            model_name='thlusermetadata',
+            index=django.contrib.postgres.indexes.GinIndex(fields=['display_name'], name='usermetadata_name_trgm', opclasses=['gin_trgm_ops']),
+        ),
+    ]
diff --git a/generalresearch/thl_django/migrations/0015_thluser_last_offerwall_requested_at_and_more.py b/generalresearch/thl_django/migrations/0015_thluser_last_offerwall_requested_at_and_more.py
new file mode 100644
index 0000000..fca705e
--- /dev/null
+++ b/generalresearch/thl_django/migrations/0015_thluser_last_offerwall_requested_at_and_more.py
@@ -0,0 +1,46 @@
+# Generated by Django 6.1.1 on 2026-10-26 18:30
+
+from django.db import migrations, models
+from django.contrib.postgres.operations import AddIndexConcurrently
+
+
+class Migration(migrations.Migration):
+    atomic = False
+
+    dependencies = [
+        ('thl_django', '0014_brokerageproduct_user_count_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='thluser',
+            name='last_offerwall_requested_at',
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.AddField(
+            model_name='thluser',
+            name='last_profiling_submission_at',
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.AddField(
+            model_name='thluser',
+            name='last_task_completed_at',
+            field=models.DateTimeField(null=True),
+        ),
+        AddIndexConcurrently(
+            model_name='thluser',
+            index=models.Index(fields=['product_id', '-last_seen'], name='thl_user_product_ceb8d3_idx'),
+        ),
+        AddIndexConcurrently(
+            model_name='thluser',
+            index=models.Index(condition=models.Q(('last_task_completed_at__isnull', False)), fields=['last_task_completed_at'], name='thl_user_last_task_comp_nn'),
+        ),
+        AddIndexConcurrently(
+            model_name='thluser',
+            index=models.Index(condition=models.Q(('last_offerwall_requested_at__isnull', False)), fields=['last_offerwall_requested_at'], name='thl_user_last_offerwall_nn'),
+        ),
+        AddIndexConcurrently(
+            model_name='thluser',
+            index=models.Index(condition=models.Q(('last_profiling_submission_at__isnull', False)), fields=['last_profiling_submission_at'], name='thl_user_last_profile_nn'),
+        ),
+    ]
diff --git a/generalresearch/thl_django/userprofile/models.py b/generalresearch/thl_django/userprofile/models.py
index 445a53c..7b4961f 100644
--- a/generalresearch/thl_django/userprofile/models.py
+++ b/generalresearch/thl_django/userprofile/models.py
@@ -88,6 +88,9 @@ class BrokerageProduct(models.Model):
     # ProductPOPFinancials model
     pop_financial = models.JSONField(default=None, null=True)
 
+    # This is the total number of users created
+    user_count = models.IntegerField(default=0, null=False)
+
     users_active_7d = models.IntegerField(default=None, null=True)
     task_completes_7d = models.IntegerField(default=None, null=True)
     # Net Earnings over the last 7 days (in USD Cents, this can be positive or negative)
diff --git a/pyproject.toml b/pyproject.toml
index 49bf634..d867224 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "generalresearch"
-version = "3.9.1"
+version = "3.9.2"
 description = "Python Utilities for General Research"
 readme = "README.md"
 requires-python = ">=3.14"