Merge branch 'main' of https://github.com/danswer-ai/danswer into bug…

…fix/beat_redux

# Conflicts:
#	backend/onyx/background/celery/tasks/indexing/tasks.py

.github/workflows/docker-build-push-model-server-container-on-tag.yml

@@ -118,6 +118,6 @@ jobs:
           TRIVY_DB_REPOSITORY: "public.ecr.aws/aquasecurity/trivy-db:2"
           TRIVY_JAVA_DB_REPOSITORY: "public.ecr.aws/aquasecurity/trivy-java-db:1"
         with:
-          image-ref: docker.io/onyxdotapp/onyx-model-server:${{ github.ref_name }}
+          image-ref: docker.io/${{ env.REGISTRY_IMAGE }}:${{ github.ref_name }}
           severity: "CRITICAL,HIGH"
           timeout: "10m"

.vscode/launch.template.jsonc

@@ -28,6 +28,7 @@
 		  		"Celery heavy", 
 		  		"Celery indexing", 
 		  		"Celery beat",
+                "Celery monitoring",
             ],
 			"presentation": {
 				 "group": "1",
@@ -51,7 +52,8 @@
 		  		"Celery light", 
 		  		"Celery heavy", 
 		  		"Celery indexing", 
-		  		"Celery beat"
+		  		"Celery beat",
+                "Celery monitoring",
 		  	],
 			"presentation": {
 				 "group": "1",
@@ -269,6 +271,31 @@
 			 },
             "consoleTitle": "Celery indexing Console"
         },
+        {
+            "name": "Celery monitoring",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "celery",
+            "cwd": "${workspaceFolder}/backend",
+            "envFile": "${workspaceFolder}/.vscode/.env",
+            "env": {},
+            "args": [
+                "-A",
+                "onyx.background.celery.versioned_apps.monitoring",
+                "worker",
+                "--pool=solo",
+                "--concurrency=1",
+                "--prefetch-multiplier=1",
+                "--loglevel=INFO",
+                "--hostname=monitoring@%n",
+                "-Q",
+                "monitoring",
+            ],
+            "presentation": {
+				 "group": "2",
+			 },
+            "consoleTitle": "Celery monitoring Console"
+        },
         {
             "name": "Celery beat",
             "type": "debugpy",

backend/alembic/versions/0f7ff6d75b57_add_index_to_index_attempt_time_created.py

@@ -0,0 +1,36 @@
+"""add index to index_attempt.time_created
+
+Revision ID: 0f7ff6d75b57
+Revises: 369644546676
+Create Date: 2025-01-10 14:01:14.067144
+
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0f7ff6d75b57"
+down_revision = "fec3db967bf7"
+branch_labels: None = None
+depends_on: None = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        op.f("ix_index_attempt_status"),
+        "index_attempt",
+        ["status"],
+        unique=False,
+    )
+
+    op.create_index(
+        op.f("ix_index_attempt_time_created"),
+        "index_attempt",
+        ["time_created"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_index_attempt_time_created"), table_name="index_attempt")
+
+    op.drop_index(op.f("ix_index_attempt_status"), table_name="index_attempt")

backend/alembic/versions/369644546676_add_composite_index_for_index_attempt_.py

@@ -0,0 +1,35 @@
+"""add composite index for index attempt time updated
+
+Revision ID: 369644546676
+Revises: 2955778aa44c
+Create Date: 2025-01-08 15:38:17.224380
+
+"""
+from alembic import op
+from sqlalchemy import text
+
+# revision identifiers, used by Alembic.
+revision = "369644546676"
+down_revision = "2955778aa44c"
+branch_labels: None = None
+depends_on: None = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_index_attempt_ccpair_search_settings_time_updated",
+        "index_attempt",
+        [
+            "connector_credential_pair_id",
+            "search_settings_id",
+            text("time_updated DESC"),
+        ],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_index_attempt_ccpair_search_settings_time_updated",
+        table_name="index_attempt",
+    )

backend/alembic/versions/97dbb53fa8c8_add_syncrecord.py

@@ -0,0 +1,72 @@
+"""Add SyncRecord
+
+Revision ID: 97dbb53fa8c8
+Revises: 369644546676
+Create Date: 2025-01-11 19:39:50.426302
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "97dbb53fa8c8"
+down_revision = "be2ab2aa50ee"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "sync_record",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("entity_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "sync_type",
+            sa.Enum(
+                "DOCUMENT_SET",
+                "USER_GROUP",
+                "CONNECTOR_DELETION",
+                name="synctype",
+                native_enum=False,
+                length=40,
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "sync_status",
+            sa.Enum(
+                "IN_PROGRESS",
+                "SUCCESS",
+                "FAILED",
+                "CANCELED",
+                name="syncstatus",
+                native_enum=False,
+                length=40,
+            ),
+            nullable=False,
+        ),
+        sa.Column("num_docs_synced", sa.Integer(), nullable=False),
+        sa.Column("sync_start_time", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("sync_end_time", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+    # Add index for fetch_latest_sync_record query
+    op.create_index(
+        "ix_sync_record_entity_id_sync_type_sync_start_time",
+        "sync_record",
+        ["entity_id", "sync_type", "sync_start_time"],
+    )
+
+    # Add index for cleanup_sync_records query
+    op.create_index(
+        "ix_sync_record_entity_id_sync_type_sync_status",
+        "sync_record",
+        ["entity_id", "sync_type", "sync_status"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_sync_record_entity_id_sync_type_sync_status")
+    op.drop_index("ix_sync_record_entity_id_sync_type_sync_start_time")
+    op.drop_table("sync_record")

backend/alembic/versions/be2ab2aa50ee_fix_capitalization.py

@@ -0,0 +1,38 @@
+"""fix_capitalization
+
+Revision ID: be2ab2aa50ee
+Revises: 369644546676
+Create Date: 2025-01-10 13:13:26.228960
+
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "be2ab2aa50ee"
+down_revision = "369644546676"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        UPDATE document
+        SET
+            external_user_group_ids = ARRAY(
+                SELECT LOWER(unnest(external_user_group_ids))
+            ),
+            last_modified = NOW()
+        WHERE
+            external_user_group_ids IS NOT NULL
+            AND external_user_group_ids::text[] <> ARRAY(
+                SELECT LOWER(unnest(external_user_group_ids))
+            )::text[]
+    """
+    )
+
+
+def downgrade() -> None:
+    # No way to cleanly persist the bad state through an upgrade/downgrade
+    # cycle, so we just pass
+    pass

backend/alembic/versions/fec3db967bf7_add_time_updated_to_usergroup_and_.py

@@ -0,0 +1,41 @@
+"""Add time_updated to UserGroup and DocumentSet
+
+Revision ID: fec3db967bf7
+Revises: 97dbb53fa8c8
+Create Date: 2025-01-12 15:49:02.289100
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "fec3db967bf7"
+down_revision = "97dbb53fa8c8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "document_set",
+        sa.Column(
+            "time_last_modified_by_user",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+    op.add_column(
+        "user_group",
+        sa.Column(
+            "time_last_modified_by_user",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("user_group", "time_last_modified_by_user")
+    op.drop_column("document_set", "time_last_modified_by_user")

backend/ee/onyx/background/celery/tasks/vespa/tasks.py

@@ -8,6 +8,9 @@
 from ee.onyx.db.user_group import mark_user_group_as_synced
 from ee.onyx.db.user_group import prepare_user_group_for_deletion
 from onyx.background.celery.apps.app_base import task_logger
+from onyx.db.enums import SyncStatus
+from onyx.db.enums import SyncType
+from onyx.db.sync_record import update_sync_record_status
 from onyx.redis.redis_usergroup import RedisUserGroup
 from onyx.utils.logger import setup_logger
 
@@ -43,24 +46,59 @@ def monitor_usergroup_taskset(
         f"User group sync progress: usergroup_id={usergroup_id} remaining={count} initial={initial_count}"
     )
     if count > 0:
+        update_sync_record_status(
+            db_session=db_session,
+            entity_id=usergroup_id,
+            sync_type=SyncType.USER_GROUP,
+            sync_status=SyncStatus.IN_PROGRESS,
+            num_docs_synced=count,
+        )
         return
 
     user_group = fetch_user_group(db_session=db_session, user_group_id=usergroup_id)
     if user_group:
         usergroup_name = user_group.name
-        if user_group.is_up_for_deletion:
-            # this prepare should have been run when the deletion was scheduled,
-            # but run it again to be sure we're ready to go
-            mark_user_group_as_synced(db_session, user_group)
-            prepare_user_group_for_deletion(db_session, usergroup_id)
-            delete_user_group(db_session=db_session, user_group=user_group)
-            task_logger.info(
-                f"Deleted usergroup: name={usergroup_name} id={usergroup_id}"
-            )
-        else:
-            mark_user_group_as_synced(db_session=db_session, user_group=user_group)
-            task_logger.info(
-                f"Synced usergroup. name={usergroup_name} id={usergroup_id}"
+        try:
+            if user_group.is_up_for_deletion:
+                # this prepare should have been run when the deletion was scheduled,
+                # but run it again to be sure we're ready to go
+                mark_user_group_as_synced(db_session, user_group)
+                prepare_user_group_for_deletion(db_session, usergroup_id)
+                delete_user_group(db_session=db_session, user_group=user_group)
+
+                update_sync_record_status(
+                    db_session=db_session,
+                    entity_id=usergroup_id,
+                    sync_type=SyncType.USER_GROUP,
+                    sync_status=SyncStatus.SUCCESS,
+                    num_docs_synced=initial_count,
+                )
+
+                task_logger.info(
+                    f"Deleted usergroup: name={usergroup_name} id={usergroup_id}"
+                )
+            else:
+                mark_user_group_as_synced(db_session=db_session, user_group=user_group)
+
+                update_sync_record_status(
+                    db_session=db_session,
+                    entity_id=usergroup_id,
+                    sync_type=SyncType.USER_GROUP,
+                    sync_status=SyncStatus.SUCCESS,
+                    num_docs_synced=initial_count,
+                )
+
+                task_logger.info(
+                    f"Synced usergroup. name={usergroup_name} id={usergroup_id}"
+                )
+        except Exception as e:
+            update_sync_record_status(
+                db_session=db_session,
+                entity_id=usergroup_id,
+                sync_type=SyncType.USER_GROUP,
+                sync_status=SyncStatus.FAILED,
+                num_docs_synced=initial_count,
             )
+            raise e
 
     rug.reset()

backend/ee/onyx/db/analytics.py

@@ -345,7 +345,8 @@ def fetch_assistant_unique_users_total(
 def user_can_view_assistant_stats(
     db_session: Session, user: User | None, assistant_id: int
 ) -> bool:
-    # If user is None, assume the user is an admin or auth is disabled
+    # If user is None and auth is disabled, assume the user is an admin
+
     if user is None or user.role == UserRole.ADMIN:
         return True