diff --git a/forum/__init__.py b/forum/__init__.py index 17007c27..186652c8 100644 --- a/forum/__init__.py +++ b/forum/__init__.py @@ -2,4 +2,4 @@ Openedx forum app. """ -__version__ = "0.6.7" +__version__ = "0.6.8" diff --git a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py index 298d47a4..81d30ac1 100644 --- a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py +++ b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py @@ -1,11 +1,17 @@ """Migration command for courses from mongodb to mysql.""" -from typing import Any +from datetime import datetime, timezone as dt_timezone +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any, cast -from django.core.management.base import BaseCommand -from django.core.management.base import CommandParser +from django.core.management.base import BaseCommand, CommandError, CommandParser +from django.db import connections +from django.utils import timezone +import forum.migration_helpers as _migration_helpers from forum.migration_helpers import ( + BATCH_SIZE, enable_mysql_backend_for_course, get_all_course_ids, migrate_content, @@ -14,12 +20,72 @@ ) from forum.mongo import get_database +# --------------------------------------------------------------------------- +# Default parallelism. +# 1 = single-threaded (safe for SQLite / test environments, default). +# N = N threads, each with its own Django DB connection; requires a +# thread-safe database backend (MySQL, PostgreSQL) for N > 1. +# --------------------------------------------------------------------------- +DEFAULT_WORKERS = 1 + + +def _migrate_one_course( + course_id: str, + create_waffle_flags: bool, + updated_since: datetime | None = None, +) -> tuple[str, float | None, str | None]: + """ + Migrate a single course in the calling thread. + + Returns ``(course_id, elapsed_seconds, error_message_or_None)``. + Each thread gets its own MongoDB client and Django DB connection. + """ + # Django stores DB connections in thread-local storage, so each worker + # thread automatically gets its own connection on first access. + # Explicitly close any connection inherited from the spawning thread + # so the worker starts with a clean slate. + connections.close_all() + + db = get_database() + t0 = time.monotonic() + try: + migrate_users(db, course_id, updated_since=updated_since) + migrate_content(db, course_id, updated_since=updated_since) + migrate_read_states(db, course_id) + if create_waffle_flags: + enable_mysql_backend_for_course(course_id) + elapsed = time.monotonic() - t0 + return course_id, elapsed, None + except Exception as exc: # pylint: disable=broad-except + elapsed = time.monotonic() - t0 + return course_id, elapsed, str(exc) + finally: + # Release the DB connection so the pool slot is returned promptly. + connections.close_all() + class Command(BaseCommand): """Migration command for courses from mongodb to mysql.""" help = "Migrate data from MongoDB to MySQL" + @staticmethod + def _parse_updated_since(raw: str | None) -> datetime | None: + """Parse an ISO8601 timestamp for incremental migration filtering.""" + if not raw: + return None + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError as exc: + raise CommandError( + "--updated-since must be a valid ISO8601 datetime, " + "for example: 2026-07-17T00:00:00Z" + ) from exc + + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed, timezone=dt_timezone.utc) + return parsed + def add_arguments(self, parser: CommandParser) -> None: """Add arguments to the command.""" parser.add_argument( @@ -28,6 +94,37 @@ def add_arguments(self, parser: CommandParser) -> None: action="store_true", help="Skip course waffle flag creation", ) + parser.add_argument( + "-w", + "--workers", + type=int, + default=DEFAULT_WORKERS, + metavar="N", + help=( + f"Number of parallel worker threads (default: {DEFAULT_WORKERS}). " + "Each worker processes one course at a time and uses its own " + "database connection. Set > 1 only with a thread-safe DB " + "backend (MySQL, PostgreSQL). Example: --workers 8" + ), + ) + parser.add_argument( + "-b", + "--batch-size", + type=int, + default=BATCH_SIZE, + metavar="N", + help=f"Bulk-operation batch size (default: {BATCH_SIZE}).", + ) + parser.add_argument( + "--updated-since", + type=str, + default=None, + help=( + "Optional ISO8601 datetime filter. Only Mongo records updated " + "on/after this time are migrated for users/content/subscriptions. " + "Example: 2026-07-17T00:00:00Z" + ), + ) parser.add_argument( "courses", nargs="+", type=str, help="List of course IDs or `all`" ) @@ -37,19 +134,90 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: db = get_database() create_waffle_flags = not options["no_toggle"] + workers: int = int(str(options["workers"])) + batch_size: int = int(str(options["batch_size"])) + updated_since_raw = cast(str | None, options.get("updated_since")) + updated_since = self._parse_updated_since(updated_since_raw) + + if workers < 1: + raise CommandError("--workers must be >= 1.") + if batch_size < 1: + raise CommandError("--batch-size must be >= 1.") + + # Override module-level BATCH_SIZE when caller passes --batch-size. + _migration_helpers.BATCH_SIZE = batch_size + course_ids = list(options["courses"]) if "all" in course_ids: course_ids = get_all_course_ids(db) - for course_id in course_ids: - self.stdout.write(f"Migrating data for course: {course_id}") - migrate_users(db, course_id) - migrate_content(db, course_id) - migrate_read_states(db, course_id) - if create_waffle_flags: - enable_mysql_backend_for_course(course_id) - self.stdout.write( - f"Enabled mysql backend waffle flag for course {course_id}." + total = len(course_ids) + run_msg = ( + f"Migrating {total} course(s) with {workers} parallel worker(s) " + f"(batch_size={_migration_helpers.BATCH_SIZE})." + ) + if updated_since is not None: + run_msg += f" updated_since={updated_since.isoformat()}" + self.stdout.write(run_msg) + + failed: list[tuple[str, str]] = [] + completed = 0 + + if workers == 1: + # Single-threaded path: simpler, no executor overhead. + for course_id in course_ids: + cid, elapsed, err = _migrate_one_course( + course_id, + create_waffle_flags, + updated_since=updated_since, ) + completed += 1 + if err: + self.stderr.write( + self.style.ERROR( + f"[{completed}/{total}] FAILED {cid} after {elapsed:.1f}s: {err}" + ) + ) + failed.append((cid, err)) + else: + self.stdout.write( + f"[{completed}/{total}] OK {cid} ({elapsed:.1f}s)" + ) + else: + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = { + pool.submit( + _migrate_one_course, + cid, + create_waffle_flags, + updated_since, + ): cid + for cid in course_ids + } + for future in as_completed(futures): + cid, elapsed, err = future.result() + completed += 1 + if err: + self.stderr.write( + self.style.ERROR( + f"[{completed}/{total}] FAILED {cid} after " + f"{elapsed:.1f}s: {err}" + ) + ) + failed.append((cid, err)) + else: + self.stdout.write( + f"[{completed}/{total}] OK {cid} ({elapsed:.1f}s)" + ) + + if failed: + self.stderr.write( + self.style.ERROR(f"\n{len(failed)} course(s) failed migration:") + ) + for cid, err in failed: + self.stderr.write(self.style.ERROR(f" {cid}: {err}")) + raise CommandError( + f"{len(failed)} course(s) failed migration. See stderr for details." + ) self.stdout.write(self.style.SUCCESS("Data migration completed successfully")) diff --git a/forum/migration_helpers.py b/forum/migration_helpers.py index 7cad1a7d..5a4c4063 100644 --- a/forum/migration_helpers.py +++ b/forum/migration_helpers.py @@ -2,10 +2,12 @@ import logging from datetime import datetime -from typing import Any +from typing import Any, cast from django.contrib.auth.models import User # pylint: disable=E5142 +from django.contrib.contenttypes.models import ContentType from django.core.management.base import OutputWrapper +from django.db.models import Max from django.utils import timezone from pymongo.collection import Collection from pymongo.database import Database @@ -28,6 +30,12 @@ logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Batch size for all bulk_create / bulk_update operations. +# Tune this to balance memory use vs round-trip overhead. +# --------------------------------------------------------------------------- +BATCH_SIZE = 500 + def get_user_or_none(user_id: Any) -> User | None: """Get a user by ID or return None if not found.""" @@ -59,448 +67,1222 @@ def get_all_course_ids(db: Database[dict[str, Any]]) -> list[str]: return db.contents.distinct("course_id") -def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate users from MongoDB to MySQL.""" - users = db.users.find({"course_stats.course_id": course_id}) - for user_data in users: - user = get_user_or_none(user_data["_id"]) - if not user: - continue +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _to_int_id(uid: Any) -> int | None: + """Convert a user ID value to int; return None on failure.""" + try: + return int(uid) + except (ValueError, TypeError): + return None + + +def _collect_all_user_ids(contents: list[dict[str, Any]]) -> set[int]: + """Return the set of all integer user IDs referenced in *contents*.""" + ids: set[int] = set() + for c in contents: + for field in ("author_id", "deleted_by", "closed_by_id"): + uid = _to_int_id(c.get(field)) + if uid is not None: + ids.add(uid) + for vote_type in ("up", "down"): + for uid_str in c.get("votes", {}).get(vote_type, []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for uid_str in c.get("abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for uid_str in c.get("historical_abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for edit in c.get("edit_history", []): + uid = _to_int_id(edit.get("author_id")) + if uid is not None: + ids.add(uid) + return ids + + +def _build_user_cache(user_ids: set[int]) -> dict[int, User]: + """Fetch all requested users in one query; return a {pk: User} mapping.""" + if not user_ids: + return {} + return {u.pk: u for u in User.objects.filter(pk__in=user_ids)} + + +def _set_if_changed(obj: Any, field: str, value: Any) -> bool: + """Set obj. only when value differs; return whether a change happened.""" + if getattr(obj, field) != value: + setattr(obj, field, value) + return True + return False + + +# --------------------------------------------------------------------------- +# migrate_users (batch-optimised) +# --------------------------------------------------------------------------- + + +def migrate_users( # pylint: disable=too-many-statements + db: Database[dict[str, Any]], + course_id: str, + updated_since: datetime | None = None, +) -> None: + """ + Migrate users from MongoDB to MySQL. + + Uses bulk_create / bulk_update instead of per-row get_or_create calls, + reducing the number of SQL round-trips from O(N) to O(1). + """ + users_query: dict[str, Any] + if updated_since is not None: + users_query = { + "course_stats": { + "$elemMatch": { + "course_id": course_id, + "last_activity_at": {"$gte": updated_since}, + } + } + } + else: + users_query = {"course_stats.course_id": course_id} + + all_mongo_users = list(db.users.find(users_query)) + if not all_mongo_users: + return + + # Build a numeric-uid → mongo-doc map, dropping unparsable IDs. + uid_map: dict[int, dict[str, Any]] = {} + for u in all_mongo_users: + uid = _to_int_id(u["_id"]) + if uid is not None: + uid_map[uid] = u + + if not uid_map: + return - _, _ = ForumUser.objects.get_or_create( - user=user, - defaults={"default_sort_key": user_data.get("default_sort_key", "date")}, + # --- Single bulk fetch of all Django users for this course --- + django_users = {u.pk: u for u in User.objects.filter(pk__in=uid_map.keys())} + + # --- ForumUser: create missing rows; update default_sort_key if changed --- + existing_fu_map: dict[int, ForumUser] = { + fu.user_id: fu # type: ignore[attr-defined] + for fu in ForumUser.objects.filter(user_id__in=django_users.keys()) + } + new_forum_users = [] + fu_to_update: list[ForumUser] = [] + for uid in django_users: + sort_key = uid_map[uid].get("default_sort_key", "date") + if uid not in existing_fu_map: + new_forum_users.append(ForumUser(user_id=uid, default_sort_key=sort_key)) + else: + fu = existing_fu_map[uid] + if _set_if_changed(fu, "default_sort_key", sort_key): + fu_to_update.append(fu) + if new_forum_users: + ForumUser.objects.bulk_create( + new_forum_users, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if fu_to_update: + ForumUser.objects.bulk_update( + fu_to_update, ["default_sort_key"], batch_size=BATCH_SIZE ) - for stat in user_data.get("course_stats", []): + # --- CourseStat: bulk create new / bulk update existing --- + existing_stats: dict[int, CourseStat] = { + cast(int, cs.user_id): cs # type: ignore[attr-defined] + for cs in CourseStat.objects.filter( + user_id__in=django_users.keys(), course_id=course_id + ) + } + stat_update_fields = [ + "active_flags", + "inactive_flags", + "threads", + "responses", + "replies", + "deleted_threads", + "deleted_responses", + "deleted_replies", + "last_activity_at", + ] + stats_to_create: list[CourseStat] = [] + stats_to_update: list[CourseStat] = [] + + for uid, mongo_user in uid_map.items(): + if uid not in django_users: + continue + for stat in mongo_user.get("course_stats", []): + if stat.get("course_id") != course_id: + continue last_activity_at = parse_mongo_datetime(stat.get("last_activity_at")) - if stat["course_id"] == course_id: - CourseStat.objects.update_or_create( - user=user, - course_id=course_id, - defaults={ - "active_flags": stat.get("active_flags", 0), - "inactive_flags": stat.get("inactive_flags", 0), - "threads": stat.get("threads", 0), - "responses": stat.get("responses", 0), - "replies": stat.get("replies", 0), - "deleted_threads": stat.get("deleted_threads", 0), - "deleted_responses": stat.get("deleted_responses", 0), - "deleted_replies": stat.get("deleted_replies", 0), - "last_activity_at": last_activity_at, - }, + if uid in existing_stats: + cs = existing_stats[uid] + stat_changed = False + stat_changed |= _set_if_changed( + cs, "active_flags", stat.get("active_flags", 0) + ) + stat_changed |= _set_if_changed( + cs, "inactive_flags", stat.get("inactive_flags", 0) + ) + stat_changed |= _set_if_changed(cs, "threads", stat.get("threads", 0)) + stat_changed |= _set_if_changed( + cs, "responses", stat.get("responses", 0) ) + stat_changed |= _set_if_changed(cs, "replies", stat.get("replies", 0)) + stat_changed |= _set_if_changed( + cs, "deleted_threads", stat.get("deleted_threads", 0) + ) + stat_changed |= _set_if_changed( + cs, "deleted_responses", stat.get("deleted_responses", 0) + ) + stat_changed |= _set_if_changed( + cs, "deleted_replies", stat.get("deleted_replies", 0) + ) + stat_changed |= _set_if_changed( + cs, "last_activity_at", last_activity_at + ) + if stat_changed: + stats_to_update.append(cs) + else: + stats_to_create.append( + CourseStat( + user_id=uid, + course_id=course_id, + active_flags=stat.get("active_flags", 0), + inactive_flags=stat.get("inactive_flags", 0), + threads=stat.get("threads", 0), + responses=stat.get("responses", 0), + replies=stat.get("replies", 0), + deleted_threads=stat.get("deleted_threads", 0), + deleted_responses=stat.get("deleted_responses", 0), + deleted_replies=stat.get("deleted_replies", 0), + last_activity_at=last_activity_at, + ) + ) + + if stats_to_create: + CourseStat.objects.bulk_create( + stats_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if stats_to_update: + CourseStat.objects.bulk_update( + stats_to_update, stat_update_fields, batch_size=BATCH_SIZE + ) -def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate content from MongoDB to MySQL.""" - contents = db.contents.find({"course_id": course_id}).sort("created_at") - for content in contents: - if content["_type"] == "CommentThread": - create_or_update_thread(content) - elif content["_type"] == "Comment": - create_or_update_comment(content) +# --------------------------------------------------------------------------- +# migrate_content (batch-optimised) +# --------------------------------------------------------------------------- - migrate_subscriptions(db, content["_id"]) +def migrate_content( + db: Database[dict[str, Any]], + course_id: str, + updated_since: datetime | None = None, +) -> None: + """ + Migrate threads and comments from MongoDB to MySQL. + + Strategy: + 1. Fetch ALL content for the course in a single MongoDB query. + 2. Build a user cache with a single Django query (eliminates the + biggest N+1 problem: per-voter/flagger/editor user lookups). + 3. Pre-fetch all existing MongoContent mappings in one query. + 4. Bulk-create new threads, then top-level comments, then child + comments (preserving the FK ordering requirement). + 5. Bulk-create votes, edit history, and abuse flaggers. + 6. Issue a SINGLE MongoDB ``$in`` query for all subscriptions + instead of one query per content item. + """ + content_query: dict[str, Any] = {"course_id": course_id} + if updated_since is not None: + content_query["updated_at"] = {"$gte": updated_since} + + contents = list(db.contents.find(content_query).sort("created_at", 1)) + + all_ids_str = [str(c["_id"]) for c in contents] + + # Pre-fetch existing MongoContent rows for selected content. + mongo_cache: dict[str, MongoContent] = {} + if all_ids_str: + mongo_cache = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) + } + + # Collect every user ID referenced in selected content; fetch them all at once. + user_cache = _build_user_cache(_collect_all_user_ids(contents)) + + # ContentType objects are cached by Django's framework after the first call. + thread_ct = ContentType.objects.get_for_model(CommentThread) + comment_ct = ContentType.objects.get_for_model(Comment) + + if contents: + threads_data = [c for c in contents if c["_type"] == "CommentThread"] + comments_data = [c for c in contents if c["_type"] == "Comment"] + + # --- Threads --- + _bulk_migrate_threads(threads_data, mongo_cache, user_cache, thread_ct) + _refresh_mongo_cache(mongo_cache, [str(t["_id"]) for t in threads_data]) + + # --- Comments: top-level first, then children (parent must exist first) --- + top_level = [ + c + for c in comments_data + if not c.get("parent_id") or str(c.get("parent_id")) == "None" + ] + child_comments = [ + c + for c in comments_data + if c.get("parent_id") and str(c.get("parent_id")) != "None" + ] + _bulk_migrate_comments(top_level, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in top_level]) + _bulk_migrate_comments(child_comments, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in child_comments]) + + # --- Metadata: votes, edit history, flaggers --- + _bulk_migrate_votes(contents, mongo_cache, user_cache, thread_ct, comment_ct) + _bulk_migrate_edit_history( + contents, mongo_cache, user_cache, thread_ct, comment_ct + ) + _bulk_migrate_abuse_flaggers( + contents, mongo_cache, user_cache, thread_ct, comment_ct + ) -def _resolve_thread_actors( - thread_data: dict[str, Any], -) -> tuple["User | None", "User | None"]: - """Return (deleted_by, closed_by) users resolved from thread_data.""" - deleted_by = ( - get_user_or_none(thread_data["deleted_by"]) - if thread_data.get("deleted_by") - else None + # --- Subscriptions: query by course (and optional updated_since) --- + _bulk_migrate_subscriptions( + db, + course_id, + mongo_cache, + user_cache, + updated_since=updated_since, ) - closed_by = ( - get_user_or_none(thread_data["closed_by_id"]) - if thread_data.get("closed_by_id") - else None + + +def _refresh_mongo_cache(cache: dict[str, MongoContent], mongo_ids: list[str]) -> None: + """Update *cache* with freshly queried MongoContent rows for *mongo_ids*.""" + if not mongo_ids: + return + cache.update( + {mc.mongo_id: mc for mc in MongoContent.objects.filter(mongo_id__in=mongo_ids)} ) - return deleted_by, closed_by -def create_or_update_thread(thread_data: dict[str, Any]) -> None: - """Create or update a thread.""" - author = get_user_or_none(thread_data["author_id"]) - if not author: +def _bulk_migrate_threads( # pylint: disable=too-many-statements + threads_data: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, +) -> None: + """Bulk-create new threads and bulk-update existing ones.""" + if not threads_data: return - # Preserve author_username from MongoDB (historical username) - author_username = thread_data.get("author_username") + new_data = [ + t + for t in threads_data + if not ( + mongo_cache.get(str(t["_id"])) + and mongo_cache[str(t["_id"])].content_object_id + ) + ] + existing_data = [ + t + for t in threads_data + if mongo_cache.get(str(t["_id"])) + and mongo_cache[str(t["_id"])].content_object_id + ] + + # --- Create new threads in bulk --- + if new_data: + pairs: list[tuple[str, CommentThread]] = [] + for t in new_data: + author = user_cache.get(_to_int_id(t.get("author_id"))) # type: ignore[arg-type] + if not author: + continue + author_username = ( + t.get("author_username") or t.get("retired_username") or author.username + ) + deleted_by = ( + user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] + if t.get("deleted_by") + else None + ) + closed_by = ( + user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] + if t.get("closed_by_id") + else None + ) + pairs.append( + ( + str(t["_id"]), + CommentThread( + author=author, + author_username=author_username, + retired_username=t.get("retired_username"), + course_id=t["course_id"], + title=get_trunc_title(t.get("title", "")), + body=t["body"], + thread_type=t.get("thread_type", "discussion"), + context=t.get("context", "course"), + anonymous=t.get("anonymous", False), + anonymous_to_peers=t.get("anonymous_to_peers", False), + closed=t.get("closed", False), + closed_by=closed_by, + close_reason_code=t.get("close_reason_code"), + pinned=t.get("pinned", False), + created_at=parse_mongo_datetime(t["created_at"]), + updated_at=parse_mongo_datetime(t["updated_at"]), + last_activity_at=parse_mongo_datetime(t["last_activity_at"]), + commentable_id=t.get("commentable_id"), + is_spam=t.get("is_spam", False), + is_deleted=t.get("is_deleted", False), + deleted_at=parse_mongo_datetime(t.get("deleted_at")), + deleted_by=deleted_by, + visible=t.get("visible", True), + ), + ) + ) - # Preserve retired_username from MongoDB - retired_username = thread_data.get("retired_username") + if pairs: + mongo_ids, objs = zip(*pairs) + obj_list = list(objs) + _course_id = obj_list[0].course_id + + # Snapshot max PK before bulk_create, then re-fetch by pk range + # narrowed to this course_id so a concurrent insert for a different + # course cannot shift the mongo_id ↔ PK zip alignment. + max_pk_before = CommentThread.objects.aggregate(Max("pk"))["pk__max"] or 0 + CommentThread.objects.bulk_create(obj_list, batch_size=BATCH_SIZE) + created = list( + CommentThread.objects.filter( + pk__gt=max_pk_before, course_id=_course_id + ).order_by("pk") + ) + if len(created) != len(obj_list): + raise RuntimeError( + f"Thread bulk_create count mismatch for course {_course_id}: " + f"submitted {len(obj_list)}, re-fetched {len(created)}. " + "A concurrent writer may have inserted rows between the PK " + "snapshot and re-fetch." + ) + mc_rows = [ + MongoContent( + mongo_id=mid, + content_type=thread_ct, + content_object_id=thread.pk, + ) + for mid, thread in zip(mongo_ids, created) + if thread.pk + ] + if mc_rows: + # Idempotent upsert for MongoContent mappings: + # - Rows that already exist with content_object_id=NULL (partial + # prior run) are updated via bulk_update. + # - Genuinely new rows are inserted via bulk_create. + # This avoids update_conflicts/unique_fields which is not + # supported on MySQL. + mc_row_map = {r.mongo_id: r for r in mc_rows} + existing_null = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter( + mongo_id__in=list(mc_row_map), + content_object_id__isnull=True, + ) + } + to_update_mc = [] + for mongo_id, existing in existing_null.items(): + new = mc_row_map[mongo_id] + existing.content_type = new.content_type # type: ignore[assignment] + existing.content_object_id = new.content_object_id + to_update_mc.append(existing) + truly_new = [r for r in mc_rows if r.mongo_id not in existing_null] + if truly_new: + MongoContent.objects.bulk_create( + truly_new, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if to_update_mc: + MongoContent.objects.bulk_update( + to_update_mc, + ["content_type", "content_object_id"], + batch_size=BATCH_SIZE, + ) + + # --- Update existing threads in bulk --- + if existing_data: + pks = [ + mongo_cache[str(t["_id"])].content_object_id + for t in existing_data + if str(t["_id"]) in mongo_cache + and mongo_cache[str(t["_id"])].content_object_id is not None + ] + thread_pk_map = {th.pk: th for th in CommentThread.objects.filter(pk__in=pks)} + thread_update_fields = [ + "title", + "body", + "thread_type", + "context", + "anonymous", + "anonymous_to_peers", + "closed", + "closed_by", + "close_reason_code", + "pinned", + "updated_at", + "last_activity_at", + "commentable_id", + "is_spam", + "is_deleted", + "deleted_at", + "deleted_by", + "visible", + ] + to_update: list[CommentThread] = [] + for t in existing_data: + mc = mongo_cache.get(str(t["_id"])) + if not mc: + continue + thread = thread_pk_map.get(mc.content_object_id) + if not thread: + continue - # If author_username is not provided, use retired_username or fallback to current username - if not author_username: - if retired_username: - author_username = retired_username - else: - author_username = author.username + updated_at = parse_mongo_datetime(t["updated_at"]) - mongo_thread_id = str(thread_data["_id"]) - mongo_content, _ = MongoContent.objects.get_or_create( - mongo_id=mongo_thread_id, - ) - if not mongo_content.content_object_id: - deleted_by, closed_by = _resolve_thread_actors(thread_data) - - thread = CommentThread.objects.create( - author=author, - author_username=author_username, - retired_username=retired_username, - course_id=thread_data["course_id"], - title=get_trunc_title(thread_data.get("title", "")), - body=thread_data["body"], - thread_type=thread_data.get("thread_type", "discussion"), - context=thread_data.get("context", "course"), - anonymous=thread_data.get("anonymous", False), - anonymous_to_peers=thread_data.get("anonymous_to_peers", False), - closed=thread_data.get("closed", False), - closed_by=closed_by, - close_reason_code=thread_data.get("close_reason_code"), - pinned=thread_data.get("pinned", False), - created_at=parse_mongo_datetime(thread_data["created_at"]), - updated_at=parse_mongo_datetime(thread_data["updated_at"]), - last_activity_at=parse_mongo_datetime(thread_data["last_activity_at"]), - commentable_id=thread_data.get("commentable_id"), - # Moderation fields - is_spam=thread_data.get("is_spam", False), - is_deleted=thread_data.get("is_deleted", False), - deleted_at=parse_mongo_datetime(thread_data.get("deleted_at")), - deleted_by=deleted_by, - visible=thread_data.get("visible", True), - ) - mongo_content.content_object_id = thread.pk - mongo_content.content_type = thread.content_type - mongo_content.save() - else: - # Update existing thread with latest data from MongoDB - thread = CommentThread.objects.get(pk=mongo_content.content_object_id) - - deleted_by, closed_by = _resolve_thread_actors(thread_data) - - # Update all fields that might have changed - thread.title = get_trunc_title(thread_data.get("title", "")) - thread.body = thread_data["body"] - thread.thread_type = thread_data.get("thread_type", "discussion") - thread.context = thread_data.get("context", "course") - thread.anonymous = thread_data.get("anonymous", False) - thread.anonymous_to_peers = thread_data.get("anonymous_to_peers", False) - thread.closed = thread_data.get("closed", False) - thread.closed_by = closed_by # type: ignore[assignment] - thread.close_reason_code = thread_data.get("close_reason_code") - thread.pinned = thread_data.get("pinned", False) - thread.updated_at = parse_mongo_datetime(thread_data["updated_at"]) # type: ignore[assignment] - thread.last_activity_at = parse_mongo_datetime(thread_data["last_activity_at"]) - thread.commentable_id = thread_data.get("commentable_id") # type: ignore[assignment] - # Update moderation fields - thread.is_spam = thread_data.get("is_spam", False) - thread.is_deleted = thread_data.get("is_deleted", False) - thread.deleted_at = parse_mongo_datetime(thread_data.get("deleted_at")) - thread.deleted_by = deleted_by # type: ignore[assignment] - thread.visible = thread_data.get("visible", True) - thread.save() - - create_or_update_edit_history(thread_data) - create_or_update_abuse_flaggers(thread_data) - create_votes(thread, thread_data.get("votes", {})) - - -def create_or_update_comment( # pylint: disable=too-many-statements - comment_data: dict[str, Any], -) -> None: - """Create or update a comment.""" - author = get_user_or_none(comment_data["author_id"]) - if not author: - return + # Fast-skip: if updated_at already matches what is in MySQL this + # thread was fully migrated in a prior run and nothing changed. + # Avoids all field comparisons and the SQL UPDATE on retry. + if thread.updated_at == updated_at: + continue - # Preserve author_username from MongoDB (historical username) - author_username = comment_data.get("author_username") + deleted_by = ( + user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] + if t.get("deleted_by") + else None + ) + closed_by = ( + user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] + if t.get("closed_by_id") + else None + ) + last_activity_at = parse_mongo_datetime(t["last_activity_at"]) + deleted_at = parse_mongo_datetime(t.get("deleted_at")) + + # Check each field directly to avoid defining a closure inside the + # loop (which captures variables by reference and confuses linters). + changed_fields: list[str] = [] + for _f, _v in ( + ("title", get_trunc_title(t.get("title", ""))), + ("body", t["body"]), + ("thread_type", t.get("thread_type", "discussion")), + ("context", t.get("context", "course")), + ("anonymous", t.get("anonymous", False)), + ("anonymous_to_peers", t.get("anonymous_to_peers", False)), + ("closed", t.get("closed", False)), + ("closed_by", closed_by), + ("close_reason_code", t.get("close_reason_code")), + ("pinned", t.get("pinned", False)), + ("updated_at", updated_at), + ("last_activity_at", last_activity_at), + ("commentable_id", t.get("commentable_id")), + ("is_spam", t.get("is_spam", False)), + ("is_deleted", t.get("is_deleted", False)), + ("deleted_at", deleted_at), + ("deleted_by", deleted_by), + ("visible", t.get("visible", True)), + ): + if _set_if_changed(thread, _f, _v): + changed_fields.append(_f) + + if changed_fields: + logger.info( + "Updated mapped thread during migration: mongo_id=%s thread_id=%s changed_fields=%s", + str(t["_id"]), + thread.pk, + ",".join(changed_fields), + ) + to_update.append(thread) + if to_update: + CommentThread.objects.bulk_update( + to_update, thread_update_fields, batch_size=BATCH_SIZE + ) - # Preserve retired_username from MongoDB - retired_username = comment_data.get("retired_username") - # If author_username is not provided, use retired_username or fallback to current username - if not author_username: - if retired_username: - author_username = retired_username - else: - author_username = author.username - - mongo_thread_id = str(comment_data["comment_thread_id"]) - mongo_thread = MongoContent.objects.filter(mongo_id=mongo_thread_id).first() - if not mongo_thread: - logger.warning( - f"Thread mapping not found for comment {comment_data.get('_id')} " - f"(mongo_thread_id={mongo_thread_id})" - ) +def _bulk_migrate_comments( # pylint: disable=too-many-statements + comments_data: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + comment_ct: ContentType, +) -> None: + """ + Bulk-create new comments and bulk-update existing ones. + + Must be called with top-level comments before child comments so that + parent PKs are available in *mongo_cache* when children are processed. + """ + if not comments_data: return - thread = CommentThread.objects.filter(pk=mongo_thread.content_object_id).first() - if not thread: - logger.warning( - f"Skipping comment {comment_data.get('_id')}: thread object not found " - f"(content_object_id={mongo_thread.content_object_id})" + + new_data = [ + c + for c in comments_data + if not ( + mongo_cache.get(str(c["_id"])) + and mongo_cache[str(c["_id"])].content_object_id ) - return - parent = None - if "parent_id" in comment_data and comment_data["parent_id"] != "None": - parent_id = str(comment_data["parent_id"]) - mongo_parent_comment = MongoContent.objects.filter(mongo_id=parent_id).first() - if not mongo_parent_comment: - logger.warning( - f"Parent mapping not found for comment {comment_data.get('_id')} " - f"(parent_id={parent_id})" + ] + existing_data = [ + c + for c in comments_data + if mongo_cache.get(str(c["_id"])) + and mongo_cache[str(c["_id"])].content_object_id + ] + + # --- Create new comments in bulk --- + if new_data: + pairs: list[tuple[str, Comment]] = [] + for c in new_data: + author = user_cache.get(_to_int_id(c.get("author_id"))) # type: ignore[arg-type] + if not author: + continue + + mongo_thread_id = str(c["comment_thread_id"]) + mc_thread = mongo_cache.get(mongo_thread_id) + if not mc_thread or not mc_thread.content_object_id: + logger.warning( + f"Thread mapping not found for comment {c.get('_id')} " + f"(mongo_thread_id={mongo_thread_id})" + ) + continue + + parent_pk: int | None = None + if c.get("parent_id") and str(c.get("parent_id")) != "None": + mc_parent = mongo_cache.get(str(c["parent_id"])) + if not mc_parent or not mc_parent.content_object_id: + logger.warning( + f"Parent mapping not found for comment {c.get('_id')} " + f"(parent_id={c['parent_id']})" + ) + continue + parent_pk = mc_parent.content_object_id + + author_username = ( + c.get("author_username") or c.get("retired_username") or author.username ) - return - parent = Comment.objects.filter( - id=mongo_parent_comment.content_object_id - ).first() - if not parent: - logger.warning( - f"Skipping comment {comment_data.get('_id')}: parent object not found " - f"(parent_content_object_id={mongo_parent_comment.content_object_id})" + deleted_by = ( + user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] + if c.get("deleted_by") + else None + ) + pairs.append( + ( + str(c["_id"]), + Comment( + author=author, + author_username=author_username, + retired_username=c.get("retired_username"), + comment_thread_id=mc_thread.content_object_id, + parent_id=parent_pk, + course_id=c["course_id"], + body=c["body"], + anonymous=c.get("anonymous", False), + anonymous_to_peers=c.get("anonymous_to_peers", False), + endorsed=c.get("endorsed", False), + child_count=c.get("child_count", 0), + created_at=parse_mongo_datetime(c["created_at"]), + updated_at=parse_mongo_datetime(c["updated_at"]), + depth=1 if parent_pk else 0, + is_spam=c.get("is_spam", False), + is_deleted=c.get("is_deleted", False), + deleted_at=parse_mongo_datetime(c.get("deleted_at")), + deleted_by=deleted_by, + visible=c.get("visible", True), + ), + ) ) - return - mongo_comment, _ = MongoContent.objects.get_or_create( - mongo_id=str(comment_data["_id"]) - ) - if not mongo_comment.content_object_id: - # Get deleted_by user if deleted_by field exists in MongoDB - deleted_by = None - if comment_data.get("deleted_by"): - deleted_by = get_user_or_none(comment_data["deleted_by"]) - - comment = Comment.objects.create( - author=author, - author_username=author_username, - retired_username=retired_username, - comment_thread=thread, - parent=parent, - course_id=comment_data["course_id"], - body=comment_data["body"], - anonymous=comment_data.get("anonymous", False), - anonymous_to_peers=comment_data.get("anonymous_to_peers", False), - endorsed=comment_data.get("endorsed", False), - child_count=comment_data.get("child_count", 0), - created_at=parse_mongo_datetime(comment_data["created_at"]), - updated_at=parse_mongo_datetime(comment_data["updated_at"]), - depth=1 if parent else 0, - # Moderation fields - is_spam=comment_data.get("is_spam", False), - is_deleted=comment_data.get("is_deleted", False), - deleted_at=parse_mongo_datetime(comment_data.get("deleted_at")), - deleted_by=deleted_by, - visible=comment_data.get("visible", True), - ) - mongo_comment.content_object_id = comment.pk - mongo_comment.content_type = comment.content_type - mongo_comment.save() - sort_key = f"{parent.pk}-{comment.pk}" if parent else f"{comment.pk}" - comment.sort_key = sort_key - comment.save() - else: - # Update existing comment with latest data from MongoDB - comment = Comment.objects.get(pk=mongo_comment.content_object_id) - - # Get deleted_by user if needed - deleted_by = None - if comment_data.get("deleted_by"): - deleted_by = get_user_or_none(comment_data["deleted_by"]) - - # Update all fields that might have changed - comment.body = comment_data["body"] - comment.anonymous = comment_data.get("anonymous", False) - comment.anonymous_to_peers = comment_data.get("anonymous_to_peers", False) - comment.endorsed = comment_data.get("endorsed", False) - comment.child_count = comment_data.get("child_count", 0) - comment.updated_at = parse_mongo_datetime(comment_data["updated_at"]) # type: ignore[assignment] - # Update moderation fields - comment.is_spam = comment_data.get("is_spam", False) - comment.is_deleted = comment_data.get("is_deleted", False) - comment.deleted_at = parse_mongo_datetime(comment_data.get("deleted_at")) - comment.deleted_by = deleted_by # type: ignore[assignment] - comment.visible = comment_data.get("visible", True) - comment.save() - - create_or_update_edit_history(comment_data) - create_or_update_abuse_flaggers(comment_data) - create_votes(comment, comment_data.get("votes", {})) - - -def create_votes(content: CommentThread | Comment, votes_data: dict[str, Any]) -> None: - """Create or update votes for a content.""" - for vote_type in ["up", "down"]: - for user_id in votes_data.get(vote_type, []): - user = get_user_or_none(user_id) - if not user: + if pairs: + mongo_ids, objs = zip(*pairs) + obj_list = list(objs) + _course_id = obj_list[0].course_id + + # Narrow re-fetch to this course so a concurrent insert for a + # different course cannot misalign the mongo_id ↔ PK zip. + max_pk_before = Comment.objects.aggregate(Max("pk"))["pk__max"] or 0 + Comment.objects.bulk_create(obj_list, batch_size=BATCH_SIZE) + created = list( + Comment.objects.filter( + pk__gt=max_pk_before, course_id=_course_id + ).order_by("pk") + ) + if len(created) != len(obj_list): + raise RuntimeError( + f"Comment bulk_create count mismatch for course {_course_id}: " + f"submitted {len(obj_list)}, re-fetched {len(created)}. " + "A concurrent writer may have inserted rows between the PK " + "snapshot and re-fetch." + ) + + mc_rows = [] + sort_key_updates: list[Comment] = [] + for mid, comment in zip(mongo_ids, created): + if not comment.pk: + continue + mc_rows.append( + MongoContent( + mongo_id=mid, + content_type=comment_ct, + content_object_id=comment.pk, + ) + ) + # Set sort_key now that we have the PK. + parent_id = cast(int | None, comment.parent_id) # type: ignore[attr-defined] + if parent_id: + comment.sort_key = f"{parent_id}-{comment.pk}" + else: + comment.sort_key = f"{comment.pk}" + sort_key_updates.append(comment) + + if mc_rows: + # Same idempotent upsert as for threads. + mc_row_map = {r.mongo_id: r for r in mc_rows} + existing_null = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter( + mongo_id__in=list(mc_row_map), + content_object_id__isnull=True, + ) + } + to_update_mc = [] + for mongo_id, existing in existing_null.items(): + new = mc_row_map[mongo_id] + existing.content_type = new.content_type # type: ignore[assignment] + existing.content_object_id = new.content_object_id + to_update_mc.append(existing) + truly_new = [r for r in mc_rows if r.mongo_id not in existing_null] + if truly_new: + MongoContent.objects.bulk_create( + truly_new, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if to_update_mc: + MongoContent.objects.bulk_update( + to_update_mc, + ["content_type", "content_object_id"], + batch_size=BATCH_SIZE, + ) + if sort_key_updates: + Comment.objects.bulk_update( + sort_key_updates, ["sort_key"], batch_size=BATCH_SIZE + ) + + # --- Update existing comments in bulk --- + if existing_data: + pks = [ + mongo_cache[str(c["_id"])].content_object_id + for c in existing_data + if str(c["_id"]) in mongo_cache + and mongo_cache[str(c["_id"])].content_object_id is not None + ] + comment_pk_map = {cm.pk: cm for cm in Comment.objects.filter(pk__in=pks)} + comment_update_fields = [ + "body", + "anonymous", + "anonymous_to_peers", + "endorsed", + "child_count", + "updated_at", + "is_spam", + "is_deleted", + "deleted_at", + "deleted_by", + "visible", + ] + to_update: list[Comment] = [] + for c in existing_data: + mc = mongo_cache.get(str(c["_id"])) + if not mc: + continue + comment: Comment | None = comment_pk_map.get(mc.content_object_id) # type: ignore[no-redef] + if not comment: continue - UserVote.objects.update_or_create( - user=user, - content_type=content.content_type, - content_object_id=content.pk, - defaults={"vote": 1 if vote_type == "up" else -1}, + updated_at = parse_mongo_datetime(c["updated_at"]) + + # Fast-skip: if updated_at already matches MySQL, nothing changed. + if comment.updated_at == updated_at: + continue + + deleted_by = ( + user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] + if c.get("deleted_by") + else None + ) + deleted_at = parse_mongo_datetime(c.get("deleted_at")) + + has_changes = False + has_changes |= _set_if_changed(comment, "body", c["body"]) + has_changes |= _set_if_changed( + comment, "anonymous", c.get("anonymous", False) + ) + has_changes |= _set_if_changed( + comment, "anonymous_to_peers", c.get("anonymous_to_peers", False) + ) + has_changes |= _set_if_changed( + comment, "endorsed", c.get("endorsed", False) + ) + has_changes |= _set_if_changed( + comment, "child_count", c.get("child_count", 0) + ) + has_changes |= _set_if_changed(comment, "updated_at", updated_at) + has_changes |= _set_if_changed(comment, "is_spam", c.get("is_spam", False)) + has_changes |= _set_if_changed( + comment, "is_deleted", c.get("is_deleted", False) + ) + has_changes |= _set_if_changed(comment, "deleted_at", deleted_at) + has_changes |= _set_if_changed(comment, "deleted_by", deleted_by) + has_changes |= _set_if_changed(comment, "visible", c.get("visible", True)) + + if has_changes: + to_update.append(comment) + if to_update: + Comment.objects.bulk_update( + to_update, comment_update_fields, batch_size=BATCH_SIZE ) -def create_or_update_edit_history(content: dict[str, Any]) -> None: - """Create or update edit history for a content.""" - edit_history = content.get("edit_history", []) - content_type = CommentThread if content["_type"] == "CommentThread" else Comment - mongo_content = MongoContent.objects.filter(mongo_id=str(content["_id"])).first() - if not mongo_content: - logger.warning( - f"Skipping edit history for content {content.get('_id')}: mapping not found" - ) +def _bulk_migrate_votes( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing UserVote rows for all content in one pass.""" + candidates: list[tuple[int, int, int, int]] = [] # (user_id, ct_id, obj_id, vote) + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: + continue + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + for vote_type in ("up", "down"): + vote_val = 1 if vote_type == "up" else -1 + for uid_str in c.get("votes", {}).get(vote_type, []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + candidates.append((uid, ct.pk, mc.content_object_id, vote_val)) + + if not candidates: return - content_object = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content_object: - logger.warning( - f"Skipping edit history for content {content.get('_id')}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + + obj_ids = {obj_id for _, _, obj_id, _ in candidates} + # Fetch existing votes WITH their direction so changed votes (e.g. up → down) + # are detected and corrected, not silently skipped. + existing_vote_map: dict[tuple[int, int, int], int] = { + (row[0], row[1], row[2]): row[3] + for row in UserVote.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id", "vote" ) - return - for edit in edit_history: - editor = get_user_or_none(edit["author_id"]) - if not editor: - continue - EditHistory.objects.get_or_create( - content_object_id=content_object.pk, - content_type=content_object.content_type, - created_at=parse_mongo_datetime(edit["created_at"]), - editor=editor, - defaults={ - "original_body": edit["original_body"], - "reason_code": edit["reason_code"], - }, + } + + new_votes: list[UserVote] = [] + changed_votes: list[tuple[int, int, int, int]] = ( + [] + ) # (uid, ct_id, obj_id, new_vote) + seen_vote_keys: set[tuple[int, int, int]] = set(existing_vote_map.keys()) + + for uid, ct_id, obj_id, vote_val in candidates: + key = (uid, ct_id, obj_id) + if key not in existing_vote_map: + if key not in seen_vote_keys: + new_votes.append( + UserVote( + user_id=uid, + content_type_id=ct_id, + content_object_id=obj_id, + vote=vote_val, + ) + ) + seen_vote_keys.add(key) + elif existing_vote_map[key] != vote_val: + # Vote direction changed between Phase 1 and Phase 2 refresh. + changed_votes.append((uid, ct_id, obj_id, vote_val)) + + if new_votes: + UserVote.objects.bulk_create( + new_votes, batch_size=BATCH_SIZE, ignore_conflicts=True ) + for uid, ct_id, obj_id, vote_val in changed_votes: + UserVote.objects.filter( + user_id=uid, content_type_id=ct_id, content_object_id=obj_id + ).update(vote=vote_val) + + +def _bulk_migrate_edit_history( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing EditHistory rows for all content in one pass.""" + obj_ids = [ + mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + ] + existing_keys: set[tuple[int, int, Any, int]] = set( + EditHistory.objects.filter(content_object_id__in=obj_ids).values_list( + "content_object_id", "content_type_id", "created_at", "editor_id" + ) + ) + to_create: list[EditHistory] = [] + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: + continue + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + for edit in c.get("edit_history", []): + editor = user_cache.get(_to_int_id(edit.get("author_id"))) # type: ignore[arg-type] + if not editor: + continue + created_at = parse_mongo_datetime(edit["created_at"]) + key = (mc.content_object_id, ct.pk, created_at, editor.pk) + if key not in existing_keys: + to_create.append( + EditHistory( + content_object_id=mc.content_object_id, + content_type=ct, + created_at=created_at, + editor=editor, + original_body=edit["original_body"], + reason_code=edit["reason_code"], + ) + ) + existing_keys.add(key) # avoid duplicates within the same batch -def create_or_update_abuse_flaggers(content: dict[str, Any]) -> None: - """Create or update abuse flaggers for content.""" - content_type = CommentThread if content["_type"] == "CommentThread" else Comment - mongo_content = MongoContent.objects.filter(mongo_id=str(content["_id"])).first() - if not mongo_content: - logger.warning( - f"Skipping abuse flaggers for content {content.get('_id')}: mapping not found" + if to_create: + EditHistory.objects.bulk_create( + to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) - return - content_object = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content_object: - logger.warning( - f"Skipping abuse flaggers for content {content.get('_id')}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + + +def _bulk_migrate_abuse_flaggers( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing AbuseFlagger / HistoricalAbuseFlagger rows.""" + obj_ids = [ + mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + ] + + existing_af: set[tuple[int, int, int]] = set( + AbuseFlagger.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id" ) - return - for user_id in content.get("abuse_flaggers", []): - user = get_user_or_none(user_id) - if not user: + ) + existing_haf: set[tuple[int, int, int]] = set( + HistoricalAbuseFlagger.objects.filter( + content_object_id__in=obj_ids + ).values_list("user_id", "content_type_id", "content_object_id") + ) + + af_to_create: list[AbuseFlagger] = [] + haf_to_create: list[HistoricalAbuseFlagger] = [] + flagged_at = timezone.now() + + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: continue - AbuseFlagger.objects.update_or_create( - user=user, - content_type=content_object.content_type, - content_object_id=content_object.pk, - defaults={ - "flagged_at": timezone.now(), - }, + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + + for uid_str in c.get("abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + key = (uid, ct.pk, mc.content_object_id) + if key not in existing_af: + af_to_create.append( + AbuseFlagger( + user_id=uid, + content_type=ct, + content_object_id=mc.content_object_id, + flagged_at=flagged_at, + ) + ) + existing_af.add(key) + + for uid_str in c.get("historical_abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + key = (uid, ct.pk, mc.content_object_id) + if key not in existing_haf: + haf_to_create.append( + HistoricalAbuseFlagger( + user_id=uid, + content_type=ct, + content_object_id=mc.content_object_id, + flagged_at=flagged_at, + ) + ) + existing_haf.add(key) + + if af_to_create: + AbuseFlagger.objects.bulk_create( + af_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) - for user_id in content.get("historical_abuse_flaggers", []): - user = get_user_or_none(user_id) - if not user: - continue - HistoricalAbuseFlagger.objects.update_or_create( - user=user, - content_type=content_object.content_type, - content_object_id=content_object.pk, - defaults={ - "flagged_at": timezone.now(), - }, + if haf_to_create: + HistoricalAbuseFlagger.objects.bulk_create( + haf_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) -def migrate_subscriptions(db: Database[dict[str, Any]], content_id: str) -> None: - """Migrate subscriptions from mongo to mysql.""" - subscriptions = db.subscriptions.find({"source_id": str(content_id)}) - for sub in subscriptions: - user = get_user_or_none(sub["subscriber_id"]) - if not user: - continue - content_type = ( - CommentThread if sub["source_type"] == "CommentThread" else Comment +def _bulk_migrate_subscriptions( + db: Database[dict[str, Any]], + course_id: str, + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + updated_since: datetime | None = None, +) -> None: + """ + Migrate subscriptions for an entire course using a SINGLE MongoDB query. + + The original code fired one ``db.subscriptions.find()`` per content + item. This version fetches them all at once via ``$in``. + """ + sub_query: dict[str, Any] = {"source.course_id": course_id} + if updated_since is not None: + sub_query["updated_at"] = {"$gte": updated_since} + + all_subs = list(db.subscriptions.find(sub_query)) + if not all_subs: + return + + thread_ct = ContentType.objects.get_for_model(CommentThread) + comment_ct = ContentType.objects.get_for_model(Comment) + + source_ids = { + str(sub.get("source_id")) + for sub in all_subs + if sub.get("source_id") is not None + } + + missing_source_ids = [sid for sid in source_ids if sid not in mongo_cache] + if missing_source_ids: + mongo_cache.update( + { + mc.mongo_id: mc + for mc in MongoContent.objects.filter(mongo_id__in=missing_source_ids) + } ) - mongo_content = MongoContent.objects.filter(mongo_id=str(content_id)).first() - if not mongo_content: - logger.warning( - f"Skipping subscription for source {content_id}: mapping not found" - ) + + mapped_source_obj_ids = [ + mc.content_object_id + for sid in source_ids + for mc in [mongo_cache.get(sid)] + if mc and mc.content_object_id + ] + + existing_sub_map: dict[tuple[int, int, int], Subscription] = { + ( + cast(int, s.subscriber_id), # type: ignore[attr-defined] + cast(int, s.source_content_type_id), # type: ignore[attr-defined] + s.source_object_id, + ): s + for s in Subscription.objects.filter(source_object_id__in=mapped_source_obj_ids) + } + + # Subscription subscribers who never authored/voted on content are absent + # from the caller's user_cache. Fetch any missing ones now. + missing_sub_uids: set[int] = set() + for sub in all_subs: + uid = _to_int_id(sub.get("subscriber_id")) + if uid is not None and uid not in user_cache: + missing_sub_uids.add(uid) + if missing_sub_uids: + extra_users = {u.pk: u for u in User.objects.filter(pk__in=missing_sub_uids)} + user_cache = {**user_cache, **extra_users} + + subs_to_create: list[Subscription] = [] + subs_to_update: list[Subscription] = [] + seen: set[tuple[int, int, int]] = set(existing_sub_map.keys()) + + for sub in all_subs: + uid = _to_int_id(sub.get("subscriber_id")) + user = user_cache.get(uid) if uid is not None else None + if not user: continue - content = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content: + mc = mongo_cache.get(str(sub.get("source_id", ""))) + if not mc or not mc.content_object_id: logger.warning( - f"Skipping subscription for source {content_id}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + f"Skipping subscription for source {sub.get('source_id')}: mapping not found" ) continue - if content: - Subscription.objects.update_or_create( - subscriber=user, - source_content_type=content.content_type, - source_object_id=content.pk, - defaults={ - "created_at": parse_mongo_datetime(sub.get("created_at")) - or timezone.now(), - "updated_at": parse_mongo_datetime(sub.get("updated_at")) - or timezone.now(), - }, + ct = thread_ct if sub.get("source_type") == "CommentThread" else comment_ct + created_at = parse_mongo_datetime(sub.get("created_at")) or timezone.now() + updated_at = parse_mongo_datetime(sub.get("updated_at")) or timezone.now() + key = (user.pk, ct.pk, mc.content_object_id) + + if key in existing_sub_map: + s = existing_sub_map[key] + # updated_at is the reliable change signal for subscriptions: + # it is set on creation and advances on any mutation. If it + # matches what is already in MySQL, nothing changed — skip. + if s.updated_at != updated_at: + s.created_at = created_at + s.updated_at = updated_at + subs_to_update.append(s) + elif key not in seen: + subs_to_create.append( + Subscription( + subscriber=user, + source_content_type=ct, + source_object_id=mc.content_object_id, + created_at=created_at, + updated_at=updated_at, + ) ) + seen.add(key) + + if subs_to_create: + Subscription.objects.bulk_create( + subs_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if subs_to_update: + Subscription.objects.bulk_update( + subs_to_update, ["created_at", "updated_at"], batch_size=BATCH_SIZE + ) + + +# --------------------------------------------------------------------------- +# migrate_read_states (batch-optimised) +# --------------------------------------------------------------------------- def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate read states from mongo to mysql.""" - users = db.users.find({"course_stats.course_id": course_id}) - for user_data in users: - user = get_user_or_none(user_data["_id"]) - if not user: + """ + Migrate read states from MongoDB to MySQL using bulk operations. + + Replaces the original per-row get_or_create / .save() loops with + one bulk-create pass for ReadState and one for LastReadTime. + """ + all_mongo_users = list(db.users.find({"course_stats.course_id": course_id})) + if not all_mongo_users: + return + + # Collect per-user read-state data and all referenced thread IDs. + all_thread_ids: set[str] = set() + user_read_data: list[tuple[int, list[dict[str, Any]]]] = [] + for user_data in all_mongo_users: + uid = _to_int_id(user_data["_id"]) + if uid is None: continue + relevant = [ + rs + for rs in user_data.get("read_states", []) + if rs.get("course_id") == course_id + ] + if relevant: + user_read_data.append((uid, relevant)) + for rs in relevant: + all_thread_ids.update(rs.get("last_read_times", {}).keys()) + + if not user_read_data: + return + + uids = {uid for uid, _ in user_read_data} + django_users = {u.pk: u for u in User.objects.filter(pk__in=uids)} - for read_state in user_data.get("read_states", []): - if read_state["course_id"] == course_id: - rs, _ = ReadState.objects.get_or_create(user=user, course_id=course_id) - for thread_id, timestamp in read_state.get( - "last_read_times", {} - ).items(): - mongo_content = MongoContent.objects.filter( - mongo_id=thread_id - ).first() - thread = mongo_content and mongo_content.content - - # For older courses using cs_comment_service, the thread may be None - # because cs_comment_service retains read_states for deleted threads - # in the users collection in MongoDB. As a result, MongoContent won't - # have a thread with deleted thread_id from read_states. - if not thread: - continue - existing_read_time = LastReadTime.objects.filter( - read_state=rs, comment_thread=thread - ).first() - if not existing_read_time: - LastReadTime.objects.create( + # Bulk-fetch all MongoContent for referenced thread IDs. + mongo_thread_map: dict[str, int] = {} # mongo_id → content_object_id + for mc in MongoContent.objects.filter(mongo_id__in=all_thread_ids): + if mc.content_object_id: + mongo_thread_map[mc.mongo_id] = mc.content_object_id + + # Bulk get-or-create ReadState rows. + existing_rs: dict[int, ReadState] = { + cast(int, rs.user_id): rs # type: ignore[attr-defined] + for rs in ReadState.objects.filter( + user_id__in=django_users.keys(), course_id=course_id + ) + } + new_rs = [ + ReadState(user=django_users[uid], course_id=course_id) + for uid, _ in user_read_data + if uid in django_users and uid not in existing_rs + ] + if new_rs: + ReadState.objects.bulk_create( + new_rs, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + existing_rs.update( + { + cast(int, rs.user_id): rs # type: ignore[attr-defined] + for rs in ReadState.objects.filter( + user__in=[r.user for r in new_rs], + course_id=course_id, + ) + } + ) + + # Bulk get-or-create LastReadTime rows. + rs_ids = [rs.pk for rs in existing_rs.values()] + existing_lrt: dict[tuple[int, int], LastReadTime] = { + ( + cast(int, lrt.read_state_id), # type: ignore[attr-defined] + cast(int, lrt.comment_thread_id), # type: ignore[attr-defined] + ): lrt + for lrt in LastReadTime.objects.filter(read_state_id__in=rs_ids) + } + + lrt_to_create: list[LastReadTime] = [] + lrt_to_update: list[LastReadTime] = [] + seen_keys: set[tuple[int, int]] = set(existing_lrt.keys()) + + for uid, read_states in user_read_data: + if uid not in django_users: + continue + rs = existing_rs.get(uid) + if not rs: + continue + for read_state in read_states: + for thread_id, timestamp in read_state.get("last_read_times", {}).items(): + thread_pk = mongo_thread_map.get(thread_id) + # thread_pk may be None for deleted threads retained in MongoDB + # read_states — skip silently (same behaviour as original code). + if not thread_pk: + continue + parsed_ts = parse_mongo_datetime(timestamp) + key = (rs.pk, thread_pk) + if key in existing_lrt: + lrt = existing_lrt[key] + if lrt.timestamp != parsed_ts: + lrt.timestamp = parsed_ts # type: ignore[assignment] + lrt_to_update.append(lrt) + elif key not in seen_keys: + lrt_to_create.append( + LastReadTime( read_state=rs, - comment_thread=thread, - timestamp=parse_mongo_datetime(timestamp), + comment_thread_id=thread_pk, + timestamp=parsed_ts, ) - else: - existing_read_time.timestamp = parse_mongo_datetime(timestamp) # type: ignore[assignment] - existing_read_time.save() + ) + seen_keys.add(key) + + if lrt_to_create: + LastReadTime.objects.bulk_create( + lrt_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if lrt_to_update: + LastReadTime.objects.bulk_update( + lrt_to_update, ["timestamp"], batch_size=BATCH_SIZE + ) def delete_course_data( @@ -510,15 +1292,14 @@ def delete_course_data( stdout: OutputWrapper, ) -> None: """Delete content (threads and comments).""" - contents = db.contents.find({"course_id": course_id}) - for content in contents: - subscriptions = ( - db.subscriptions.delete_many({"source_id": str(content["_id"])}) - if not dry_run - else None - ) + contents = list(db.contents.find({"course_id": course_id})) + content_ids = [str(c["_id"]) for c in contents] + # Delete all subscriptions in one $in query instead of one-per-content. + sub_result = None + if content_ids and not dry_run: + sub_result = db.subscriptions.delete_many({"source_id": {"$in": content_ids}}) stdout.write( - f"Subscription documents to be deleted: {subscriptions.deleted_count if subscriptions else 'N/A (dry run)'}" + f"Subscription documents to be deleted: {sub_result.deleted_count if sub_result else 'N/A (dry run)'}" ) content_result = ( @@ -575,6 +1356,8 @@ def enable_mysql_backend_for_course(course_id: str) -> None: from forum.toggles import ENABLE_MYSQL_BACKEND course_key = CourseKey.from_string(course_id) - WaffleFlagCourseOverrideModel.objects.create( - course_id=course_key, waffle_flag=ENABLE_MYSQL_BACKEND.name, enabled=True + WaffleFlagCourseOverrideModel.objects.update_or_create( + course_id=course_key, + waffle_flag=ENABLE_MYSQL_BACKEND.name, + defaults={"enabled": True}, ) diff --git a/tests/test_management/test_commands/test_migration_commands.py b/tests/test_management/test_commands/test_migration_commands.py index 06965897..3b8b915c 100644 --- a/tests/test_management/test_commands/test_migration_commands.py +++ b/tests/test_management/test_commands/test_migration_commands.py @@ -1,5 +1,6 @@ """Test forum mongodb migration commands.""" +from datetime import timedelta from io import StringIO from typing import Any @@ -23,7 +24,6 @@ ) from forum.utils import get_trunc_title - pytestmark = pytest.mark.django_db @@ -946,3 +946,292 @@ def test_migrate_comment_fallback_to_current_username( mongo_comment = MongoContent.objects.get(mongo_id=comment_id) comment = Comment.objects.get(pk=mongo_comment.content_object_id) assert comment.author_username == "current_username" + + +def test_remigrate_updates_changed_thread_and_comment( + patched_mongodb: Database[Any], +) -> None: + """Ensure remigration updates already-mapped thread/comment when Mongo changed.""" + thread_id = ObjectId() + comment_id = ObjectId() + first_time = timezone.now() + second_time = first_time + timedelta(minutes=5) + + patched_mongodb.contents.insert_many( + [ + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Initial Title", + "body": "Initial thread body", + "created_at": first_time, + "updated_at": first_time, + "last_activity_at": first_time, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + { + "_id": comment_id, + "_type": "Comment", + "author_id": "1", + "course_id": "test_course", + "body": "Initial comment body", + "created_at": first_time, + "updated_at": first_time, + "comment_thread_id": thread_id, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + "depth": 0, + "sk": f"{comment_id}", + }, + ] + ) + + User.objects.create(id=1, username="testuser") + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + patched_mongodb.contents.update_one( + {"_id": thread_id}, + { + "$set": { + "title": "Updated Title", + "body": "Updated thread body", + "updated_at": second_time, + "last_activity_at": second_time, + } + }, + ) + patched_mongodb.contents.update_one( + {"_id": comment_id}, + { + "$set": { + "body": "Updated comment body", + "updated_at": second_time, + } + }, + ) + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + mongo_thread = MongoContent.objects.get(mongo_id=thread_id) + migrated_thread = CommentThread.objects.get(pk=mongo_thread.content_object_id) + assert migrated_thread.title == "Updated Title" + assert migrated_thread.body == "Updated thread body" + + mongo_comment = MongoContent.objects.get(mongo_id=comment_id) + migrated_comment = Comment.objects.get(pk=mongo_comment.content_object_id) + assert migrated_comment.body == "Updated comment body" + + +def test_remigrate_skips_noop_thread_bulk_update( + patched_mongodb: Database[Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """Ensure unchanged mapped threads are not sent to bulk_update.""" + thread_id = ObjectId() + now = timezone.now() + + patched_mongodb.contents.insert_one( + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Stable Title", + "body": "Stable body", + "created_at": now, + "updated_at": now, + "last_activity_at": now, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + } + ) + User.objects.create(id=1, username="testuser") + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + original_bulk_update = CommentThread.objects.bulk_update + bulk_update_calls: list[int] = [] + + def _tracking_bulk_update( + objs: list[CommentThread], + fields: list[str], + batch_size: int | None = None, + ) -> None: + bulk_update_calls.append(len(objs)) + original_bulk_update(objs, fields, batch_size=batch_size) + + monkeypatch.setattr(CommentThread.objects, "bulk_update", _tracking_bulk_update) + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + assert not bulk_update_calls + + +def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: + """Only content updated on/after cutoff should be migrated.""" + old_thread_id = ObjectId() + new_thread_id = ObjectId() + now = timezone.now() + cutoff = now - timedelta(hours=1) + + User.objects.create(id=1, username="testuser") + + patched_mongodb.contents.insert_many( + [ + { + "_id": old_thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Old thread", + "body": "Old body", + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), + "last_activity_at": now - timedelta(days=2), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + { + "_id": new_thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "New thread", + "body": "New body", + "created_at": now - timedelta(minutes=30), + "updated_at": now - timedelta(minutes=30), + "last_activity_at": now - timedelta(minutes=30), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + assert not MongoContent.objects.filter(mongo_id=str(old_thread_id)).exists() + assert MongoContent.objects.filter(mongo_id=str(new_thread_id)).exists() + + +def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: + """Only users with recent course_stats.last_activity_at should be migrated.""" + now = timezone.now() + cutoff = now - timedelta(hours=1) + + User.objects.create(id=1, username="olduser") + User.objects.create(id=2, username="newuser") + + patched_mongodb.users.insert_many( + [ + { + "_id": "1", + "username": "olduser", + "default_sort_key": "date", + "course_stats": [ + { + "course_id": "test_course", + "threads": 1, + "last_activity_at": now - timedelta(days=2), + } + ], + }, + { + "_id": "2", + "username": "newuser", + "default_sort_key": "date", + "course_stats": [ + { + "course_id": "test_course", + "threads": 2, + "last_activity_at": now - timedelta(minutes=10), + } + ], + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + assert not CourseStat.objects.filter(user_id=1, course_id="test_course").exists() + assert CourseStat.objects.filter(user_id=2, course_id="test_course").exists() + + +def test_updated_since_filters_subscriptions(patched_mongodb: Database[Any]) -> None: + """Subscription delta should use subscription.updated_at, even if content is older.""" + now = timezone.now() + cutoff = now - timedelta(hours=1) + thread_id = ObjectId() + + User.objects.create(id=1, username="author") + User.objects.create(id=2, username="sub_old") + User.objects.create(id=3, username="sub_new") + + patched_mongodb.contents.insert_one( + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Thread", + "body": "Body", + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), + "last_activity_at": now - timedelta(days=2), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + } + ) + + # First run creates MongoContent mapping for the thread. + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + patched_mongodb.subscriptions.insert_many( + [ + { + "subscriber_id": "2", + "source_id": str(thread_id), + "source_type": "CommentThread", + "source": {"course_id": "test_course"}, + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), + }, + { + "subscriber_id": "3", + "source_id": str(thread_id), + "source_type": "CommentThread", + "source": {"course_id": "test_course"}, + "created_at": now - timedelta(minutes=30), + "updated_at": now - timedelta(minutes=30), + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + subs = Subscription.objects.all() + assert subs.count() == 1 + assert subs.first().subscriber_id == 3 # type: ignore[union-attr]